#include "SortTests.h"

CSortTests::CSortTests(void):m_psRel1Buffer(NULL), m_psRel2Buffer(NULL), m_nMaxElemInList(0), m_nMaxElemMul(0), m_nBufferLimitedByteCount(0)
{

	QueryPerformanceFrequency((LARGE_INTEGER*)&m_freq);

}	

CSortTests::~CSortTests(void)
{
}

// Description: in this sort function we are sorting in log(N) iterations. N is the number of sorted blocks - we are sorting in the
//					beginning and then we sort every 2^i blocks until we get 2^i=N
template <class T>
bool CSortTests::SortInIterations(int nCurrIteration, int nTotalNumOfIterations, wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
											int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, wstring& rwsFinalFileName, wstring &rwsAppend)
{
	

	/////////////////////////////////////////////////////////////////////
	/// Left TODO:	1. Handle the case of uneven buffers num (for example 5) - we have
	///					the formula but we didn't implement it.
	///				2. Fix a bug: every run we handle 2 sets of size nNumOfBlocksInIteration / 2 V
	///				3. Check if it is working on small and big relations
	///				4. Check what happen if the amount is smaller then sector (512)
	/////////////////////////////////////////////////////////////////////

	if (0 == nCurrIteration)
	{
		// This is the first iteration, we should sort block by block
		CreateBlockSortedRelation<T>( wsRelFileName, wsSortedFileName, wsSortedFileName, nRelRowsNum,
			nRelByteCount, rfTimeSec, pfIoTimeMsec, rnReadSectorsNum, rnWrittenSectorsNum, false);
		rwsFinalFileName = wsSortedFileName;	
	}
	else
	{
		// Sort every 2^nCurrIteration blocks
		int nNumOfBlocksInIteration		= static_cast<int>(pow(2.0f, nCurrIteration));
		int nNumOfBlocksInLastIteration	= static_cast<int>(pow(2.0f, nCurrIteration-1));
		int nEveryBufferByteCount		= static_cast<int>(m_nBufferLimitedByteCount / 2.0f);//static_cast<float>(nNumOfBlocksInIteration));
		int nEveryBufferRowsNum			= (nEveryBufferByteCount / sizeof(T));
		int nNumOfBlocks				=  static_cast<int>(ceil( static_cast<float>(nRelByteCount) / m_nBufferLimitedByteCount));
		int nNumOfRuns					= static_cast<int>(ceil(nNumOfBlocks / static_cast<float>(nNumOfBlocksInIteration)));
		int nNumOfLeftBlocks			= (nNumOfBlocks % nNumOfBlocksInIteration);
		int nResBlockNumOfRows			= (m_nBufferLimitedByteCount/ sizeof(T));

		T* psRelBuffer				= NULL;
		T* psRelSoretdBuffer		= NULL;

		// TODO: handle when nEveryBufferByteCount is less then a sector
		if (nEveryBufferByteCount < 512 || 0 != (nEveryBufferByteCount % 512))
		{
			printf("SortInIterations - Problem - not 512 aligned\n");
			return false;
		}

		if (AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount ) == true)
		{
			if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount ) != true)
			{
				DeAllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount);
				return false;
			}
		}


		Disk* psDiskRel			= new Disk(wsSortedFileName, eLogical, eRead);

		wsSortedFileName.append(rwsAppend);
		
		Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

		SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(2/*nNumOfBlocksInIteration*/ * sizeof(SSortedBufferInfo<T>));

		if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
		{
			if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
			{
				DWORDLONG dwlOffset			= 0;
				DWORDLONG dwlSoretdOffset	= 0;
				int nSortedBufferIndex		= 0;

				LARGE_INTEGER  startTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

				for (int i = 0; i < nNumOfRuns; ++i)
				{
					/*printf("Run %d \n", i);*/
					int nFirstBufferSize	= nNumOfBlocksInLastIteration;
					int nSecondBufferSize	= nNumOfBlocksInIteration;
					
					bool bLeftOver = false;
					if ((nNumOfRuns - 1 == i) && (0  != nNumOfLeftBlocks))
					{
						// in the last iteration we have to merge the "leftovers"
						/*printf("Must merge out of the regular merge\n");*/
						
						// We have to merge some blocks that we didn't merge in the regular 
						// merge because their number was too small.
						// for example: we have 7 blocks and we merged every 1, 2, 4, 8 blocks. 
						// We didn't handle the last block when we merge every 2, but we have to
						// merge the last 3 blocks when we merge every 4 blocks. Then when we merge
						// every 8 blocks we will merge the first 4 blocks with the last 3 blocks.

						if (nNumOfLeftBlocks < nNumOfBlocksInIteration && nNumOfLeftBlocks > nNumOfBlocksInLastIteration)
						{
							// We can merge this block with the blocks that were merge last iteration

							nFirstBufferSize	= nNumOfBlocksInLastIteration;
							nSecondBufferSize	= nNumOfLeftBlocks - nFirstBufferSize;

							/*printf("merge block size %d with block size %d \n", nFirstBufferSize, nSecondBufferSize);*/
							bLeftOver = true;

						}
						else
						{
							//printf("Don't merge %d blocks in iteration %d with %d num of blocks in iteration \n",
							//	nNumOfLeftBlocks, nCurrIteration, nNumOfBlocksInIteration);
		
							for (int j = 0; j < nNumOfLeftBlocks; ++j)
							{
								dwlOffset = (j+i*nNumOfBlocksInIteration) * m_nBufferLimitedByteCount;

								if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
								{
									if (false == psDiskSortedRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
									{
										break;
									}
									dwlSoretdOffset += m_nBufferLimitedByteCount;
								}
							}
							break;
						}
					}

					// Read for the first time - from evey block nEveryBufferByteCount
					for (int j = 0; j < 2/*nNumOfBlocksInIteration*/; ++j)
					{
						dwlOffset	= ((j*nNumOfBlocksInLastIteration + i*nNumOfBlocksInIteration) * m_nBufferLimitedByteCount);
						int nIndex	= (j*nEveryBufferRowsNum);
											
						if (psDiskRel->DoOverlappedIO(psRelBuffer+nIndex, nEveryBufferByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
						{
							psBuffers[j].m_nTotalReadBytes	= nEveryBufferByteCount / 512;
							psBuffers[j].m_nLeftBytesToRead = (((nNumOfBlocksInIteration/2)*m_nBufferLimitedByteCount) -
																	nEveryBufferByteCount);
							psBuffers[j].m_pBuffer			= psRelBuffer+nIndex;	
							psBuffers[j].m_nCurrIndex		= 0;
							psBuffers[j].m_dwOffset			= dwlOffset;	
							psBuffers[j].m_bIsFinished		= false;

							if ((true == bLeftOver) && (1==j))
							{
								// This is the second buffer - its not whole
								psBuffers[j].m_nLeftBytesToRead = ((nSecondBufferSize*m_nBufferLimitedByteCount) - 
																	nEveryBufferByteCount);

							}

							/*printf("Read buffer %d - offset %d\n", j, dwlOffset);*/

							//for (int k = 0; k < nEveryBufferByteCount/sizeof(T); ++k)
							//{
							//	printf("%d,", psBuffers[j].m_pBuffer[k]);
							//}
							//printf("\n");
						}
						else
						{
							printf("SortInIterations - Problem - failed reading\n");
							break;
						}
					}
					
					bool bRes = false;

					// Perform the merge
					while(bRes == false)
					{
						// Get the smaller element of whole buffers
						int nSmallerIndex = 0;
						//for (int j = 0; j < 2; ++j)
						//{
						if (false == psBuffers[1].m_bIsFinished && false == psBuffers[nSmallerIndex].m_bIsFinished)
						{
							if (*(psBuffers[nSmallerIndex].m_pBuffer + psBuffers[nSmallerIndex].m_nCurrIndex) > 
								*(psBuffers[1].m_pBuffer + psBuffers[1].m_nCurrIndex))
							{
								nSmallerIndex = 1;
							}
						}
						else if (false == psBuffers[1].m_bIsFinished)
						{
							nSmallerIndex = 1;
						}
						//}
						if (false == psBuffers[nSmallerIndex].m_bIsFinished)
						{
							// Write the smaller element to the result buffer
							psRelSoretdBuffer[nSortedBufferIndex++] = 
								*(psBuffers[nSmallerIndex].m_pBuffer + psBuffers[nSmallerIndex].m_nCurrIndex);

							if(nSortedBufferIndex >= 2 && psRelSoretdBuffer[nSortedBufferIndex-1] <  psRelSoretdBuffer[nSortedBufferIndex-2])
							{
								printf("found - %d - %d\n", psRelSoretdBuffer[nSortedBufferIndex-1].GetKey(), psRelSoretdBuffer[nSortedBufferIndex-2].GetKey());

							}
							psBuffers[nSmallerIndex].m_nCurrIndex++;

						}

						// We have reached to the end of a reading buffer
						if (psBuffers[nSmallerIndex].m_nCurrIndex >= nEveryBufferRowsNum && 
							(false == psBuffers[nSmallerIndex].m_bIsFinished))
						{
							//if (m_nBufferLimitedByteCount == (psBuffers[nSmallerIndex].m_dwOffset - (nSmallerIndex + i) * m_nBufferLimitedByteCount))
							//{
							//	// There is no need to read more - this is the end of this block
							//	psBuffers[nSmallerIndex].m_bIsFinished = true;
							//}
							//else 
							if (0 >= psBuffers[nSmallerIndex].m_nLeftBytesToRead)
							{
								// We have read the whole buffer
								psBuffers[nSmallerIndex].m_bIsFinished = true;
							}
							else
							{
								// We have to read one more buffer
								dwlOffset = psBuffers[nSmallerIndex].m_dwOffset + nEveryBufferByteCount;

								int nReadByteCount = min(nEveryBufferByteCount, psBuffers[nSmallerIndex].m_nLeftBytesToRead);

								if (psDiskRel->DoOverlappedIO(psBuffers[nSmallerIndex].m_pBuffer, nReadByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
								{
									psBuffers[nSmallerIndex].m_nTotalReadBytes += nReadByteCount / 512;
									//rnReadSectorsNum += nEveryBufferByteCount / 512;
									psBuffers[nSmallerIndex].m_nCurrIndex = 0;
									psBuffers[nSmallerIndex].m_dwOffset = dwlOffset;
									psBuffers[nSmallerIndex].m_nLeftBytesToRead -= nReadByteCount;

									/*printf("Read buffer %d - offset %d\n", nSmallerIndex, dwlOffset);*/
									//for (int k = 0; k < nReadByteCount/sizeof(T); ++k)
									//{
									//	printf("%d,", psBuffers[nSmallerIndex].m_pBuffer[k]);
									//}
									//printf("\n");

								}
								else
								{
									printf("SortInIterations - Problem - failed reading\n");
									break;
								}
							}

						}

						// We have reached to the end of the results buffer - we have to write the result to the disk
						if (nResBlockNumOfRows <= nSortedBufferIndex)
						{
							if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, m_nBufferLimitedByteCount, dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
							{
								printf("SortInIterations - Problem - failed reading\n");
								break;
							}
							/*printf("************Wrote to disk to offset %d \n", dwlSoretdOffset);*/
							dwlSoretdOffset += m_nBufferLimitedByteCount;
							nSortedBufferIndex = 0;
						}

						if ((true == psBuffers[0].m_bIsFinished) && (true == psBuffers[1].m_bIsFinished))
						{
							bRes = true;
						}
					}

//					DWORD dwEnd = timeGetTime();

				}

				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);


			}
			psDiskRel->CloseDisk();
			psDiskSortedRel->CloseDisk();
		}

		//rnReadSectorsNum += psBuffers[0].m_nTotalReadBytes;
		//rnReadSectorsNum += psBuffers[1].m_nTotalReadBytes;
		free(psBuffers);
		delete(psDiskRel);
		delete(psDiskSortedRel);
		
		DeAllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount);
		DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount);

				

	}



	if (nCurrIteration >= nTotalNumOfIterations-1)
	{
		// Finish
		rwsFinalFileName = wsSortedFileName;
		return true;
	}
	else
	{
		// Proceed with the iterations
		SortInIterations<T>(nCurrIteration+1, nTotalNumOfIterations, wsRelFileName, wsSortedFileName, nRelRowsNum,
			nRelByteCount, rfTimeSec, pfIoTimeMsec, rnReadSectorsNum, rnWrittenSectorsNum, rwsFinalFileName, rwsAppend);
	}
	return true;
	
}

// Description: Suppose the memory size is M and the number of blocks of the relation is N. then every time we will sort together M/N 
//					elements from every block - in this way we are reading the relation only twice and write one intermediate results
//					and one final results.
template <class T>
bool CSortTests::SortInBlocks(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
											int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, wstring wsAppend)
{
	////////////////////////////////////////////////////////////////////////////////////////////////
	// For start we should sort block by block
	CreateBlockSortedRelation<T>( wsRelFileName, wsSortedFileName, wsSortedFileName, nRelRowsNum,
		nRelByteCount, rfTimeSec, pfIoTimeMsec, rnReadSectorsNum, rnWrittenSectorsNum, false);

	printf("CreateBlockSortedRelation took %f msec and IO time was %f \n", rfTimeSec, *pfIoTimeMsec); 
	///////////////////////////////////////////////////////////////////////////////////////////////
	// Start the merge of the sorted blocks to a one sorted file
	int m_nBufferLimitedRowsCount		= (m_nBufferLimitedByteCount/sizeof(T));
	
	int nReadBufferBytesCount			= (m_nBufferLimitedByteCount/2);

	int nBuffersLimitedRowsNum			= (nReadBufferBytesCount / sizeof(T));

	int nWriteBufferBytesCount			= (m_nBufferLimitedByteCount/2);

	int nWriteBuffersLimitedRowsNum		= (nWriteBufferBytesCount / sizeof(T));

	// number of blocks is the relation size / memory limit - the same like in CreateBlockSortedRelation  
	int nNumOfBlocks				= static_cast<int>(ceil((static_cast<float>(nRelByteCount)) / m_nBufferLimitedByteCount));

	// we will read from every sorted block M/N bytes
	int nEveryBufferByteCount		= static_cast<int>(ceil(nReadBufferBytesCount / static_cast<float>(nNumOfBlocks)));

	// we will read from every sorted block M/N rows
	int nEveryBufferRowsNum			= (nEveryBufferByteCount / sizeof(T));

	// we will need x iterations in order to read the whole block
	int nNumOfRuns					= static_cast<int>(ceil(m_nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));

	int nLastBufferByteCount		= nEveryBufferByteCount;

	int nLastBufferRowsNum			= nEveryBufferRowsNum;

	// TODO: check if the case in which there is one last block which its size is smaller than others.

	// we need N buffers in order to read them together
	T** psRelBuffer				= (T**)malloc(nNumOfBlocks*sizeof(T*));
	T* psRelSoretdBuffer		= NULL;

	// TODO: handle when nEveryBufferByteCount is less then a sector



	//////////////////////////////////////////////////////////
	int nEveryBufferSecorsCount = (nEveryBufferByteCount / 512);

	if (0 == nEveryBufferSecorsCount)
	{
		// This will not happen because CreateBlockSortedRelation succeeded
		printf("Shouldn't have happened\n");
	}
	else
	{
		// Every buffer will be aligned to 512  - the last one will be bigger than the others
		nEveryBufferByteCount	= (nEveryBufferSecorsCount * 512);

		int nRemainder			= (nReadBufferBytesCount - (nEveryBufferByteCount*(nNumOfBlocks-1)));

		if ( nRemainder > nEveryBufferByteCount)
		{
			nLastBufferByteCount = nRemainder;

			if (0 != (nRemainder%512))
			{
				printf("Problem = 0 != (nRemaider%512) \n");
			}
		}
		else
		{
			nLastBufferByteCount = nEveryBufferByteCount;
		}

		nLastBufferRowsNum		= (nLastBufferByteCount/sizeof(T));

		nEveryBufferRowsNum		= (nEveryBufferByteCount/sizeof(T));

	}
	//////////////////////////////////////////////////////////////

	bool bRes = true;
	int i = 0;

	// Allocate virtual memory for the buffers
	for (i = 0; i < nNumOfBlocks; ++i)
	{
		int nByteCount = nEveryBufferByteCount;
		if(i == (nNumOfBlocks-1))
		{
			nByteCount = nLastBufferByteCount;
		}
		if (AllocateVirtualBuffer((void**)(&(psRelBuffer[i])), nByteCount ) != true)
		{
			bRes = false;
			break;
		}
	}

	if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount ) != true)
	{
		bRes = false;		
	}

	if (false == bRes)
	{
		// Deallocate the previous buffers that were successfully been allocated
		// TODO: check if we have deallocate until i or until i-1
		for (int j = 0; j < i; ++j)
		{
			int nByteCount = nEveryBufferByteCount;
			if(j == (nNumOfBlocks-1))
			{
				nByteCount = nLastBufferByteCount;
			}
			DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
		}

		return false;
	}


	// This disk is for reding the intermediate results of the sorted buffers.
	Disk* psDiskRel			= new Disk(wsSortedFileName, eLogical, eRead);
	
	// The results file name will be wsSortedFileName + 1
	wsSortedFileName.append(wsAppend);
	Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

	// Allocating the meta-data structure for everyone of the buffers
	SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(nNumOfBlocks * sizeof(SSortedBufferInfo<T>));

	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlSoretdOffset	= 0;
			int nSortedBufferIndex		= 0;

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			int nByteCount = nEveryBufferByteCount;

			// Read the first part of every buffer
			for (int j = 0; j < nNumOfBlocks; ++j)
			{
				if (j == (nNumOfBlocks - 1))
				{
					// This is the last buffer
					nByteCount = nLastBufferByteCount;
				}

				dwlOffset	= (j * m_nBufferLimitedByteCount);

				int nIndex	= (j * nEveryBufferRowsNum);
				
				// Read the data
				if (psDiskRel->DoOverlappedIO((psRelBuffer[j]), nByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					// Update the meta-data
					psBuffers[j].m_nTotalReadBytes	= nByteCount;

					psBuffers[j].m_nLeftBytesToRead	= (m_nBufferLimitedByteCount - nByteCount);

					psBuffers[j].m_nCurrIndex = 0;

					psBuffers[j].m_dwOffset = dwlOffset;	

					psBuffers[j].m_nBytesCount = nByteCount;

					psBuffers[j].m_nRowsCount  = (nByteCount/sizeof(T));

					// what if there is only one part to read
					if(0 >= psBuffers[j].m_nLeftBytesToRead)
					{
						psBuffers[j].m_bIsFinished = true;
					}
					else
					{
						psBuffers[j].m_bIsFinished = false;
					}

					/*printf("Read buffer %d - offset %d\n", j, dwlOffset);*/

					//for (int k = 0; k < nEveryBufferByteCount/sizeof(T); ++k)
					//{
					//	printf("%d,", psBuffers[j].m_pBuffer[k]);
					//}
					//printf("\n");
				}
				else
				{
					printf("SortInIterations - Problem - failed reading\n");
					break;
				}
			}


			// Perform the sorting
			//for (int i = 0; i < nNumOfRuns; ++i)
			//{
//					printf("Run %d \n", i);
				
			bool bRes = true;

			while(bRes == true)
			{
				bRes = false;

				T* sElem = NULL;

				int nSmallerBufferIndex = 0;
				
				T* sMinElem = NULL;

				int nFirstBuffer = 0;
				// Get the first buffer that we didn't finish reading to be the first smaller one
				for (int j = 0; j < nNumOfBlocks; ++j)
				{
					if (false == psBuffers[j].m_bIsFinished)
					{
						// The first buffere which we didn't finish reading
						sMinElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						nFirstBuffer = j;
						nSmallerBufferIndex = j;
						bRes = true;
						break;
					}

				}				

				// Get the smaller element of whole buffers
				for (int j = nFirstBuffer+1; j < nNumOfBlocks; ++j)
				{
					if (false == psBuffers[j].m_bIsFinished)
					{
						sElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						
						if (*sElem < *sMinElem)
						{
							// We have smaller element
							nSmallerBufferIndex = j;
							sMinElem = sElem;
						}

						// We still have buffers that we didn't finish reading
						bRes = true;
					}
				}

				if (true == bRes)
				{
					// Write the smaller element to the results buffer
					psRelSoretdBuffer[nSortedBufferIndex++] = *(sMinElem);

					// Increase the current index
					psBuffers[nSmallerBufferIndex].m_nCurrIndex++;

					//int nByteCount = nEveryBufferByteCount;
					//int nRowsCount = nEveryBufferRowsNum;

					//if ((nNumOfBlocks - 1) == nSmallerBufferIndex)
					//{
					//	nByteCount = nLastBufferByteCount;
					//	nRowsCount = nLastBufferRowsNum;

					//}

					// We have reached to the end of this part - must read another part
					if (psBuffers[nSmallerBufferIndex].m_nRowsCount <= psBuffers[nSmallerBufferIndex].m_nCurrIndex)
					{
						if (0 >= psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead)
						{
							// We have read the whole buffer
							psBuffers[nSmallerBufferIndex].m_bIsFinished = true;
						}
						else
						{
							// We have to read one more part
							dwlOffset = psBuffers[nSmallerBufferIndex].m_dwOffset + psBuffers[nSmallerBufferIndex].m_nBytesCount;

							// Maybe the left bytes to read is smaller than the part size
							int nReadByteCount = min(psBuffers[nSmallerBufferIndex].m_nBytesCount, psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead);

							if (nReadByteCount != psBuffers[nSmallerBufferIndex].m_nBytesCount)
							{
								psBuffers[nSmallerBufferIndex].m_nBytesCount	= nReadByteCount;
								psBuffers[nSmallerBufferIndex].m_nRowsCount		= (nReadByteCount/sizeof(T));
							}
							//if (nReadByteCount < nByteCount)
							//{
							//	printf("bla\n");
							//}
		
							if (psDiskRel->DoOverlappedIO((psRelBuffer[nSmallerBufferIndex]), nReadByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
							{
								psBuffers[nSmallerBufferIndex].m_nTotalReadBytes	+= nReadByteCount;
								psBuffers[nSmallerBufferIndex].m_nCurrIndex			= 0;
								psBuffers[nSmallerBufferIndex].m_dwOffset			= dwlOffset;
								psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead	-= nReadByteCount;

								if (psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead < 0)
								{
									printf("nnnnnnn\n");
								}
								psBuffers[nSmallerBufferIndex].m_bIsFinished		= false;

								/*printf("Read buffer %d - offset %d\n", nSmallerIndex, dwlOffset);*/
								//for (int k = 0; k < nReadByteCount/sizeof(T); ++k)
								//{
								//	printf("%d,", psBuffers[nSmallerIndex].m_pBuffer[k]);
								//}
								//printf("\n");

							}
							else
							{
								printf("SortInIterations - Problem - failed reading\n");
								break;
							}
						}
					}
				}

				// If we have reached to the end of the results buffer or we finished reading all blocks 
				//		we have to write the results to the disk
				if ((nWriteBuffersLimitedRowsNum <= nSortedBufferIndex || false == bRes) && (nSortedBufferIndex > 0))
				{
					if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, nSortedBufferIndex*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						printf("SortInIterations - Problem - failed writing\n");
						break;
					}

					dwlSoretdOffset += nSortedBufferIndex*sizeof(T);
					nSortedBufferIndex = 0;
				}

			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

		}

	}
	psDiskRel->CloseDisk();
	psDiskSortedRel->CloseDisk();
	//}

	free(psBuffers);
	delete(psDiskRel);
	delete(psDiskSortedRel);
	
	// Deallocate the memory for the read-buffers
	for (int j = 0; j < nNumOfBlocks; ++j)
	{
		int nByteCount = nEveryBufferByteCount;
		if(j == (nNumOfBlocks-1))
		{
			nByteCount = nLastBufferByteCount;
		}

		DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
	}

	// Deallocate the memory for the result buffer
	DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount);	

	return bRes;

}


template <class T>
bool CSortTests::SortInBlocksV1(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
											int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, 
											int& rnReadSectorsNum, int& rnWrittenSectorsNum, 
											wstring wsAppend, double fReadBufferRatio)
{
	////////////////////////////////////////////////////////////////////////////////////////////////
	// For start we should sort block by block
	CreateBlockSortedRelation<T>( wsRelFileName, wsSortedFileName, wsSortedFileName, nRelRowsNum,
		nRelByteCount, rfTimeSec, pfIoTimeMsec, rnReadSectorsNum, rnWrittenSectorsNum, false);

//	printf("CreateBlockSortedRelation took %f msec and IO time was %f \n", rfTimeSec, *pfIoTimeMsec); 
	///////////////////////////////////////////////////////////////////////////////////////////////
	// Start the merge of the sorted blocks to a one sorted file
	// number of blocks is the relation size / memory limit - the same like in CreateBlockSortedRelation  
	int nNumOfBlocks = static_cast<int>(ceil((static_cast<float>(nRelByteCount)) / m_nBufferLimitedByteCount));

	//int nMemLimitedSectorsCount			= (m_nBufferLimitedByteCount/512);

	//double nReadSectorsNum				= ((nMemLimitedSectorsCount/2.0)*fReadBufferRatio);

	//int nEveryBufferSectorsCount		= static_cast<int>(ceil(nMemLimitedSectorsCount/nReadSectorsNum));

//	printf("number of sectors in every read buffer is %d \n",nEveryBufferSectorsCount); 
	//
	//int nReadBufferBytesCount			= static_cast<int>(nEveryBufferSectorsCount*512*nNumOfBlocks);

	int nReadBufferBytesCount			= static_cast<int>((m_nBufferLimitedByteCount/2)*fReadBufferRatio);

	int nReadBufferLimitedRowsNum		= (nReadBufferBytesCount / sizeof(T));

	int nWriteBufferBytesCount			= (m_nBufferLimitedByteCount - nReadBufferBytesCount);

	int nWriteBuffersLimitedRowsNum		= (nWriteBufferBytesCount / sizeof(T));

	// we will read from every sorted block M/N bytes
	int nEveryBufferByteCount		= static_cast<int>(ceil(nReadBufferBytesCount / static_cast<float>(nNumOfBlocks)));

	// TODO: if nEveryBufferByteCount is not sector aligned


	// we will read from every sorted block M/N rows
	int nEveryBufferRowsNum			= (nEveryBufferByteCount / sizeof(T));

	// we will need x iterations in order to read the whole block
	int nNumOfRuns					= static_cast<int>(ceil(m_nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));

	int nLastBufferByteCount		= nEveryBufferByteCount;

	int nLastBufferRowsNum			= nEveryBufferRowsNum;

	// TODO: check if the case in which there is one last block which its size is smaller than others.

	// we need N buffers in order to read them together
	T** psRelBuffer				= (T**)malloc(nNumOfBlocks*sizeof(T*));
	T* psRelSoretdBuffer		= NULL;

	// TODO: handle when nEveryBufferByteCount is less then a sector



	//////////////////////////////////////////////////////////
	int nEveryBufferSecorsCount = (nEveryBufferByteCount / 512);

	if (0 == nEveryBufferSecorsCount)
	{
		// This will not happen because CreateBlockSortedRelation succeeded
		printf("Shouldn't have happened\n");
	}
	else
	{
		// Every buffer will be aligned to 512  - the last one will be bigger than the others
		nEveryBufferByteCount	= (nEveryBufferSecorsCount * 512);

		int nRemainder			= (nReadBufferBytesCount - (nEveryBufferByteCount*(nNumOfBlocks-1)));

		if ( nRemainder > nEveryBufferByteCount)
		{
			nLastBufferByteCount = nRemainder;

			if (0 != (nRemainder%512))
			{
				printf("Problem = 0 != (nRemaider%512) \n");
			}
		}
		else
		{
			nLastBufferByteCount = nEveryBufferByteCount;
		}

		nLastBufferRowsNum		= (nLastBufferByteCount/sizeof(T));

		nEveryBufferRowsNum		= (nEveryBufferByteCount/sizeof(T));

	}

	printf("memory size is: %d, Read buffer: %d, last read buffer: %d, write buffer:%d\n", m_nBufferLimitedByteCount, nEveryBufferByteCount, nLastBufferByteCount, nWriteBufferBytesCount); 
	//////////////////////////////////////////////////////////////

	bool bRes = true;
	int i = 0;

	// Allocate virtual memory for the buffers
	for (i = 0; i < nNumOfBlocks; ++i)
	{
		int nByteCount = nEveryBufferByteCount;
		if(i == (nNumOfBlocks-1))
		{
			nByteCount = nLastBufferByteCount;
		}
		if (AllocateVirtualBuffer((void**)(&(psRelBuffer[i])), nByteCount ) != true)
		{
			bRes = false;
			break;
		}
	}

	if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount ) != true)
	{
		bRes = false;		
	}

	if (false == bRes)
	{
		// Deallocate the previous buffers that were successfully been allocated
		// TODO: check if we have deallocate until i or until i-1
		for (int j = 0; j < i; ++j)
		{
			int nByteCount = nEveryBufferByteCount;
			if(j == (nNumOfBlocks-1))
			{
				nByteCount = nLastBufferByteCount;
			}
			DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
		}

		return false;
	}


	// This disk is for reding the intermediate results of the sorted buffers.
	Disk* psDiskRel			= new Disk(wsSortedFileName, eLogical, eRead);
	
	// The results file name will be wsSortedFileName + 1
	wsSortedFileName.append(wsAppend);
	Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

	// Allocating the meta-data structure for everyone of the buffers
	SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(nNumOfBlocks * sizeof(SSortedBufferInfo<T>));

	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlSoretdOffset	= 0;
			int nSortedBufferIndex		= 0;

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			int nByteCount = nEveryBufferByteCount;

			// Read the first part of every buffer
			for (int j = 0; j < nNumOfBlocks; ++j)
			{
				if (j == (nNumOfBlocks - 1))
				{
					// This is the last buffer
					nByteCount = nLastBufferByteCount;
				}

				dwlOffset	= (j * m_nBufferLimitedByteCount);

				int nIndex	= (j * nEveryBufferRowsNum);
				
				// Read the data
				if (psDiskRel->DoOverlappedIO((psRelBuffer[j]), nByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					// Update the meta-data
					psBuffers[j].m_nTotalReadBytes	= nByteCount;

					psBuffers[j].m_nLeftBytesToRead	= (m_nBufferLimitedByteCount - nByteCount);

					psBuffers[j].m_nCurrIndex = 0;

					psBuffers[j].m_dwOffset = dwlOffset;	

					psBuffers[j].m_nBytesCount = nByteCount;

					psBuffers[j].m_nRowsCount  = (nByteCount/sizeof(T));

					// what if there is only one part to read
					if(0 >= psBuffers[j].m_nLeftBytesToRead)
					{
						psBuffers[j].m_bIsFinished = true;
					}
					else
					{
						psBuffers[j].m_bIsFinished = false;
					}

					/*printf("Read buffer %d - offset %d\n", j, dwlOffset);*/

					//for (int k = 0; k < nEveryBufferByteCount/sizeof(T); ++k)
					//{
					//	printf("%d,", psBuffers[j].m_pBuffer[k]);
					//}
					//printf("\n");
				}
				else
				{
					printf("SortInIterations - Problem - failed reading\n");
					break;
				}
			}


			// Perform the sorting
			//for (int i = 0; i < nNumOfRuns; ++i)
			//{
//					printf("Run %d \n", i);
				
			bool bRes = true;

			while(bRes == true)
			{
				bRes = false;

				T* sElem = NULL;

				int nSmallerBufferIndex = 0;
				
				T* sMinElem = NULL;

				int nFirstBuffer = 0;
				// Get the first buffer that we didn't finish reading to be the first smaller one
				for (int j = 0; j < nNumOfBlocks; ++j)
				{
					if (false == psBuffers[j].m_bIsFinished)
					{
						// The first buffere which we didn't finish reading
						sMinElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						nFirstBuffer = j;
						nSmallerBufferIndex = j;
						bRes = true;
						break;
					}

				}				

				// Get the smaller element of whole buffers
				for (int j = nFirstBuffer+1; j < nNumOfBlocks; ++j)
				{
					if (false == psBuffers[j].m_bIsFinished)
					{
						sElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						
						if (*sElem < *sMinElem)
						{
							// We have smaller element
							nSmallerBufferIndex = j;
							sMinElem = sElem;
						}

						// We still have buffers that we didn't finish reading
						bRes = true;
					}
				}

				if (true == bRes)
				{
					// Write the smaller element to the results buffer
					psRelSoretdBuffer[nSortedBufferIndex++] = *(sMinElem);

					// Increase the current index
					psBuffers[nSmallerBufferIndex].m_nCurrIndex++;

					//int nByteCount = nEveryBufferByteCount;
					//int nRowsCount = nEveryBufferRowsNum;

					//if ((nNumOfBlocks - 1) == nSmallerBufferIndex)
					//{
					//	nByteCount = nLastBufferByteCount;
					//	nRowsCount = nLastBufferRowsNum;

					//}

					// We have reached to the end of this part - must read another part
					if (psBuffers[nSmallerBufferIndex].m_nRowsCount <= psBuffers[nSmallerBufferIndex].m_nCurrIndex)
					{
						if (0 >= psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead)
						{
							// We have read the whole buffer
							psBuffers[nSmallerBufferIndex].m_bIsFinished = true;
						}
						else
						{
							// We have to read one more part
							dwlOffset = psBuffers[nSmallerBufferIndex].m_dwOffset + psBuffers[nSmallerBufferIndex].m_nBytesCount;

							// Maybe the left bytes to read is smaller than the part size
							int nReadByteCount = min(psBuffers[nSmallerBufferIndex].m_nBytesCount, psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead);

							if (nReadByteCount != psBuffers[nSmallerBufferIndex].m_nBytesCount)
							{
								psBuffers[nSmallerBufferIndex].m_nBytesCount	= nReadByteCount;
								psBuffers[nSmallerBufferIndex].m_nRowsCount		= (nReadByteCount/sizeof(T));
							}
							//if (nReadByteCount < nByteCount)
							//{
							//	printf("bla\n");
							//}
		
							if (psDiskRel->DoOverlappedIO((psRelBuffer[nSmallerBufferIndex]), nReadByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
							{
								psBuffers[nSmallerBufferIndex].m_nTotalReadBytes	+= nReadByteCount;
								psBuffers[nSmallerBufferIndex].m_nCurrIndex			= 0;
								psBuffers[nSmallerBufferIndex].m_dwOffset			= dwlOffset;
								psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead	-= nReadByteCount;

								if (psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead < 0)
								{
									printf("nnnnnnn\n");
								}
								psBuffers[nSmallerBufferIndex].m_bIsFinished		= false;

								/*printf("Read buffer %d - offset %d\n", nSmallerIndex, dwlOffset);*/
								//for (int k = 0; k < nReadByteCount/sizeof(T); ++k)
								//{
								//	printf("%d,", psBuffers[nSmallerIndex].m_pBuffer[k]);
								//}
								//printf("\n");

							}
							else
							{
								printf("SortInIterations - Problem - failed reading\n");
								break;
							}
						}
					}
				}

				// If we have reached to the end of the results buffer or we finished reading all blocks 
				//		we have to write the results to the disk
				if ((nWriteBuffersLimitedRowsNum <= nSortedBufferIndex || false == bRes) && (nSortedBufferIndex > 0))
				{
					if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, nSortedBufferIndex*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						printf("SortInIterations - Problem - failed writing\n");
						break;
					}

					dwlSoretdOffset += nSortedBufferIndex*sizeof(T);
					nSortedBufferIndex = 0;
				}

			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

		}

	}
	psDiskRel->CloseDisk();
	psDiskSortedRel->CloseDisk();
	//}

	free(psBuffers);
	delete(psDiskRel);
	delete(psDiskSortedRel);
	
	// Deallocate the memory for the read-buffers
	for (int j = 0; j < nNumOfBlocks; ++j)
	{
		int nByteCount = nEveryBufferByteCount;
		if(j == (nNumOfBlocks-1))
		{
			nByteCount = nLastBufferByteCount;
		}

		DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
	}

	// Deallocate the memory for the result buffer
	DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer), m_nBufferLimitedByteCount);	

	return bRes;

}


// Description: Read the whole relation and keep the k smallest elements. Then write them and read the relation again and keep the 
//	next k smallest elements, and so on.
//	We will write the relation once and read the relation logk(M) times.

template <class T>
bool CSortTests::MergeSortKsV1(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
{
	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList	= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul		= 1;

	n_nLastBufferElems	= 0;

	// the results buffer and the read buffer are the same size.
	int nBuffersLimitedRowsNum		= (m_nBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(m_nBufferLimitedByteCount)));

	unsigned int  nMaxResultSize	= nBuffersLimitedRowsNum;
	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
	int nNumOfRuns					= nNumberOfBuffersToRead;//static_cast<int>(ceil(m_nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));

	// Allocate buffers

	T* psRelBuffer				= NULL;
	T* psRelResultBuffer		= NULL;
	T* psResultBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount ) == true)
	{
		// Handle two results buffer 
		if (AllocateVirtualBuffer((void**)(&psResultBuffer), 2*m_nBufferLimitedByteCount ) != true)
		{
			DeAllocateVirtualBuffer((void**)(&psRelBuffer), 2*m_nBufferLimitedByteCount);
			return false;
		}
	}

	psRelResultBuffer	= psResultBuffer;
	int nResultOffset	= 0;
	int nResIndex		= 0;

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			m_nMaxElemInList	= -1;
			m_nMaxElemMul		= 0;

			for (int j = 0; j < nNumOfRuns; ++j)
			{
				psRelResultBuffer	= psResultBuffer;
				nResultOffset		= nBuffersLimitedRowsNum;
				dwlOffset			= 0;
				n_nLastBufferElems	= 0;

				// 1. Read the first buffer of the relation

				if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
				{		
					dwlOffset+= m_nBufferLimitedByteCount;

					// 1.1 Sort the first buffer of the relation
					QuickSort(psRelBuffer, 0, nBuffersLimitedRowsNum);

					// 1.3 Copy it to the result buffer
					memcpy(psRelResultBuffer, psRelBuffer, m_nBufferLimitedByteCount);

					n_nLastBufferElems = nBuffersLimitedRowsNum;

					// 2. For the next buffers of the relation

					for (int i = 0; i < nNumberOfBuffersToRead - 1; ++i)
					{
						// 2.1 Read the buffer
						if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
						{		
							dwlOffset+= m_nBufferLimitedByteCount;

							// 2.2 Sort the buffer
							QuickSort(psRelBuffer, 0, nBuffersLimitedRowsNum);

//							if (psRelBuffer[0] <= psRelResultBuffer[nBuffersLimitedRowsNum-1])
							{
								// 2.3 Merge the new buffer with the last one and take the smallest M
								
								int nIndex1 = 0;
								int nIndex2 = 0;
								nResIndex	= 0;


								// Perform a merge sort between the new buffer and the result buffer
								// In order not to copy a lot of elements we will have another buffer of pointers and
								// in the end we will copy them to the result buffer.

								T** aBuffer = (T**)malloc(nBuffersLimitedRowsNum*sizeof(T*));

								int nCurrElemMul = 0;

								// TODO: handle the case in which the first buffer was not full because
								//			there were not enough elements bigger then the last maximum

								// 2.3.1 Iterate over the two buffers
								while (	nIndex1 < nBuffersLimitedRowsNum && 
										nIndex2 < n_nLastBufferElems && 
										nResIndex < nBuffersLimitedRowsNum )
								{
									if ((m_nMaxElemInList > psRelBuffer[nIndex1].GetKey()))
									{
										// the elemets are smaller then the last maximum number
										// they already been handle
										nIndex1++;
									}
									else if ((m_nMaxElemInList > psRelResultBuffer[nIndex2].GetKey()))
									{
										// the elemets are smaller then the last maximum number
										// they already been handle
										nIndex2++;
									}
									else if (m_nMaxElemInList == psRelResultBuffer[nIndex2].GetKey() && nCurrElemMul < m_nMaxElemMul)
									{
										nCurrElemMul++;
										nIndex2++;
									}
									else if (m_nMaxElemInList == psRelBuffer[nIndex1].GetKey() && nCurrElemMul < m_nMaxElemMul)
									{
										nCurrElemMul++;
										nIndex1++;
									}

									else
									{
										if (psRelBuffer[nIndex1] < psRelResultBuffer[nIndex2])
										{
											aBuffer[nResIndex] = &(psRelBuffer[nIndex1]);
											++nResIndex;
											++nIndex1;
										}
										else
										{
											aBuffer[nResIndex] = &(psRelResultBuffer[nIndex2]);
											++nResIndex;
											++nIndex2;
										}
									} // if ((m_nMaxElemInList > psRelBuffer[nIndex1].GetKey()))

								} // while (	nIndex1 < nBuffersLimitedRowsNum && 

								// 2.3.2 Copy the rest of the list (if one of the lists were over)
								if (nResIndex < nBuffersLimitedRowsNum)
								{
									if (nIndex2 < n_nLastBufferElems)
									{
										while (nResIndex < nBuffersLimitedRowsNum && nIndex2 < n_nLastBufferElems)
										{
											aBuffer[nResIndex] = &(psRelResultBuffer[nIndex2]);
											++nResIndex;
											++nIndex2;
										}
									}
									if (nIndex1 < nBuffersLimitedRowsNum)
									{
										while (nResIndex < nBuffersLimitedRowsNum && nIndex1 < nBuffersLimitedRowsNum)
										{
											aBuffer[nResIndex] = &(psRelBuffer[nIndex1]);
											++nResIndex;
											++nIndex1;
										}
									}
	
								} // if (nResIndex < nBuffersLimitedRowsNum)

								n_nLastBufferElems = nResIndex;
								printf("%d. Num of elements in result buffer = %d \n", j, nResIndex);
								//printf("%d,%d - switched psRelResultBuffer to %d offset\n", j,i, nResultOffset);
								// 2.3.3 Switch the result buffer pointer
								psRelResultBuffer = psResultBuffer + nResultOffset;

								if (0 == nResultOffset)
								{
									nResultOffset = nBuffersLimitedRowsNum;
								}
								else
								{
									nResultOffset = 0;
								}
								
								// 2.3.4 Copy the pointers into the results buffer

								for (nIndex1 = 0; nIndex1 < nResIndex;  ++nIndex1)
								{
									psRelResultBuffer[nIndex1] = *(aBuffer[nIndex1]);
								}
							} // if (psRelBuffer[0] <= psRelResultBuffer[nBuffersLimitedRowsNum-1])

						} // if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
						else
						{
							// the sorted new buffer first element is already bigger then the first
							// element of the result buffer - no merge is needed.
						}
					} // for (int i = 0; i < nNumberOfBuffersToRead - 1; ++i)


				} // if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))

				
				// 4.1 Get the maximum number of this buffer and its multiplicity
				
				if (nResIndex != nBuffersLimitedRowsNum)
				{
					printf("Problem\n");
				}
				m_nMaxElemInList	= psRelResultBuffer[nBuffersLimitedRowsNum-1].GetKey();
				m_nMaxElemMul		= 0;

				for (int i = nBuffersLimitedRowsNum-1; i >= 0; --i)
				{
					if (psRelResultBuffer[i].GetKey() == m_nMaxElemInList)
					{
						m_nMaxElemMul++;
					}
					else
					{
						break;
					}
				}

//				printf("%d - wrote the buffer to offset %d\n", j,dwlResultOffset);
				// 4.2 Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psRelResultBuffer, m_nBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum))
				{							
					dwlResultOffset += m_nBufferLimitedByteCount;
				}

			} //for (int j = 0; j < nNumOfRuns; ++j)

		} // 	if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())


	} // 	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())

	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount);

	DeAllocateVirtualBuffer((void**)(&psResultBuffer), 2*m_nBufferLimitedByteCount);

	return bRes;



}


template <class T>
bool CSortTests::MergeSortKsV2(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
{
	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList	= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul		= 1;

	// We devide the memory to read buffer and write buffer
	int nHalfBufferLimitedByteCount		= (m_nBufferLimitedByteCount/2);
	// the results buffer and the read buffer are the same size.
	int nBuffersLimitedRowsNum			= (nHalfBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuffersToRead			= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nHalfBufferLimitedByteCount)));

	unsigned int  nMaxResultSize		= nBuffersLimitedRowsNum;
	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
	int nNumOfRuns						= nNumberOfBuffersToRead;//static_cast<int>(ceil(m_nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));

	// Allocate buffers

	T* psRelBuffer				= NULL;
	T* psResultBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psResultBuffer), m_nBufferLimitedByteCount ) == false)
	{
		return false;
	}

	psRelBuffer	= psResultBuffer;

	int nResIndex		= 0;
	int nIndex			= 0;
	int nCurrMaxMul		= 0;
	int nLastItrElems	= 0;

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			m_nMaxElemInList	= -1;
			m_nMaxElemMul		= 0;

			for (int j = 0; j < nNumOfRuns; ++j)
			{			
				int nBufferOffset	= 0;
				int nNumOfElements	= 0;
				nCurrMaxMul			= 0;
				nLastItrElems		= 0;
				dwlOffset			= 0;

				for (int i = 0; i < nNumberOfBuffersToRead; ++i)
				{
					psRelBuffer	= psResultBuffer + nBufferOffset;
							
					// 1. Read the first buffer of the relation

					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nHalfBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
					{		
						dwlOffset+= nHalfBufferLimitedByteCount;

						nResIndex	= 0;
						nIndex		= 0;

						// Take only the relevant elements.
						for (nIndex = 0; nIndex < nBuffersLimitedRowsNum; ++nIndex)
						{
							if (psRelBuffer[nIndex].GetKey() < m_nMaxElemInList)
							{
								continue;
							}

							if (psRelBuffer[nIndex].GetKey() == m_nMaxElemInList && nCurrMaxMul < m_nMaxElemMul)
							{
								nCurrMaxMul++;
								continue;
							}
							
							psRelBuffer[nResIndex] = psRelBuffer[nIndex];
							++nResIndex;


						}


					}

					// 1.1 Sort the buffer of the relation
					QuickSort2(psResultBuffer, 0, nLastItrElems + nResIndex-1);

					nLastItrElems	= min(nLastItrElems + nResIndex, nBuffersLimitedRowsNum);

					nBufferOffset	= nLastItrElems;

					psRelBuffer		= psResultBuffer + nBufferOffset;

				}

				// 4.1 Get the maximum number of this buffer and its multiplicity
				
				m_nMaxElemInList	= psResultBuffer[nBuffersLimitedRowsNum-1].GetKey();
				m_nMaxElemMul		= 0;

				for (int i = nNumOfElements-1; i > 0; --i)
				{
					if (psResultBuffer[i].GetKey() == m_nMaxElemInList)
					{
						m_nMaxElemMul++;
					}
					else
					{
						break;
					}
				}


				if (nLastItrElems < nBuffersLimitedRowsNum)
				{
					printf("CSortTests:: the buffer to write is not full \n");
				}

				// 4.2 Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psResultBuffer, nHalfBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
				{							
					dwlResultOffset += nHalfBufferLimitedByteCount;
				}
			}

				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);
		}
	}


	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psResultBuffer), m_nBufferLimitedByteCount);

	return bRes;
}

// This function is different from MergeSortKsV2 in: the buffers are not half and half devided. you can choose
//	the ratio between them.
template <class T>
bool CSortTests::MergeSortKsV3(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, 
					int& rnWrittenSectorsNum, double fReadBufferRatio, bool bUseQuickSort /*= true*/)
{
	int nHalfBufferLimitedByteCount = (m_nBufferLimitedByteCount/2);

	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList	= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul		= 1;

	int nSectorByteCount	= static_cast<int>(nHalfBufferLimitedByteCount*fReadBufferRatio);

	int nLastSectorByteCount = nSectorByteCount;

	int nSectorRowsNum		= (nSectorByteCount/sizeof(T));

	// the results buffer and the read buffer are the same size.
	int nBuffersLimitedRowsNum		= (nHalfBufferLimitedByteCount / sizeof(T));

	// We will read one sector at a time
	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nSectorByteCount)));

	if (nRelByteCount%nSectorByteCount)
	{
		nLastSectorByteCount = nRelByteCount%nSectorByteCount;
	}

	int  nMaxResultSize				= (2*nHalfBufferLimitedByteCount - nSectorByteCount);

	int nResultMaxRowsCount			= (nMaxResultSize/sizeof(T));

//	printf("the read buffer is %d , the write buffer is %d \n", nSectorByteCount, nMaxResultSize);

	
	int nNumOfRuns					= static_cast<int>(ceil(nRelByteCount/static_cast<float>(nMaxResultSize)));

	// Allocate buffers

	T* psRelBuffer				= NULL;
	T* psResultBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psResultBuffer), 2*nHalfBufferLimitedByteCount ) == false)
	{
		return false;
	}

	memset(psResultBuffer, 0 , 2*nHalfBufferLimitedByteCount);

	psRelBuffer	= psResultBuffer;

	int nResIndex		= 0;
	int nIndex			= 0;
	int nCurrMaxMul		= 0;
	int nLastItrElems	= 0;

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			m_nMaxElemInList	= -1;
			m_nMaxElemMul		= 0;

			for (int j = 0; j < nNumOfRuns; ++j)
			{		
				if (nNumOfRuns-1 == j)
				{
					nMaxResultSize		= (nRelByteCount - j*nMaxResultSize);
					nResultMaxRowsCount = (nMaxResultSize/ sizeof(T));
				}
				int nBufferOffset	= 0;
				int nNumOfElements	= 0;
				nCurrMaxMul			= 0;
				nLastItrElems		= 0;
				dwlOffset			= 0;

				for (int i = 0; i < nNumberOfBuffersToRead; ++i)
				{
					int nReadByteCount	= nSectorByteCount;
					int nRowsNum		= nSectorRowsNum;

					if (nNumberOfBuffersToRead-1 == i)
					{
						nReadByteCount	= nLastSectorByteCount;
						nRowsNum		= nLastSectorByteCount/sizeof(T);
					}
					psRelBuffer	= psResultBuffer + nBufferOffset;
							
					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nReadByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
					{		
						dwlOffset+= nReadByteCount;

						nResIndex	= 0;
						nIndex		= 0;

						// Take only the relevant elements.
						for (nIndex = 0; nIndex < nRowsNum; ++nIndex)
						{
							if (psRelBuffer[nIndex].GetKey() < m_nMaxElemInList)
							{
								continue;
							}

							if (psRelBuffer[nIndex].GetKey() == m_nMaxElemInList && nCurrMaxMul < m_nMaxElemMul)
							{
								nCurrMaxMul++;
								continue;
							}

							
							psRelBuffer[nResIndex] = psRelBuffer[nIndex];
							++nResIndex;
						}
					}
					else
					{
						printf("MergeSortKsV3: Problem Reading\n");
					}
			
					// 1.1 Sort the buffer of the relation

					if (true == bUseQuickSort)
					{
						// Use simple quick sort in orede to merge the two buffers,
						QuickSort2(psResultBuffer, 0, nLastItrElems + nResIndex-1, false);
						nLastItrElems	= min(nLastItrElems + nResIndex, nResultMaxRowsCount);
					}
					else
					{
						// Use merge 2 sorted buffers sort
						nLastItrElems = Merge2SortedArrays<SRel2>(psResultBuffer, nLastItrElems, nResIndex, nResultMaxRowsCount);
						nLastItrElems = min(nLastItrElems, nResultMaxRowsCount);
					}

					nBufferOffset	= nLastItrElems;

					psRelBuffer		= psResultBuffer + nBufferOffset;

				}

				// 4.1 Get the maximum number of this buffer and its multiplicity
		
				m_nMaxElemInList	= psResultBuffer[nLastItrElems-1].GetKey();

				m_nMaxElemMul = 0;
				for (int i = nLastItrElems-1; i > 0; --i)
				{
					if (psResultBuffer[i].GetKey() == m_nMaxElemInList)
					{
						m_nMaxElemMul++;
					}
					else
					{
						break;
					}
				}

				// 4.2 Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psResultBuffer, nMaxResultSize, dwlResultOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
				{							
					dwlResultOffset += nMaxResultSize;
				}
			}

				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);
		}
	}


	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psResultBuffer), 2*nHalfBufferLimitedByteCount);

	return bRes;
}


template <class T>
bool CSortTests::MergeSortKsV5(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, double fReadBufferRatio)
{
	// In this version, we will read one sector and will write M-1 sectors
	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList			= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul				= 1;

	int nWriteSectorByteCount	= static_cast<int>(m_nBufferLimitedByteCount*(2-fReadBufferRatio));

	float fEraseUnitBytes		= (32.0f*512.0f);

	nWriteSectorByteCount		= static_cast<int>((ceilf(nWriteSectorByteCount/fEraseUnitBytes))*fEraseUnitBytes);

	if (0 == nWriteSectorByteCount)
	{
		printf("shouldn't have happened \n");
		return false;
	}

	int nReadSectorByteCount	= static_cast<int>(2*m_nBufferLimitedByteCount - nWriteSectorByteCount);

	printf("the read buffer is %d , the write buffer is %d \n", nReadSectorByteCount, nWriteSectorByteCount);

	int nLastSectorByteCount	= nReadSectorByteCount;

	int nSectorRowsNum			= (nReadSectorByteCount/sizeof(T));

	// We will read one sector at a time
	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(nReadSectorByteCount)));

	if (0 != (nRelByteCount%nReadSectorByteCount))
	{
		nLastSectorByteCount = nRelByteCount%nReadSectorByteCount;
	}

	int  nMaxResultSize				= nWriteSectorByteCount;

	int nResultMaxRowsCount			= (nMaxResultSize/sizeof(T));
	
	int nNumOfRuns					= static_cast<int>(ceil(nRelByteCount/static_cast<float>(nMaxResultSize)));

	// Allocate buffers

	T* psRelBuffer				= NULL;
	T* psResultBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psResultBuffer), 2*m_nBufferLimitedByteCount ) == false)
	{
		return false;
	}

	memset(psResultBuffer, 0 , 2*m_nBufferLimitedByteCount);

	psRelBuffer	= psResultBuffer;

	int nResIndex		= 0;
	int nIndex			= 0;
	int nCurrMaxMul		= 0;
	int nLastItrElems	= 0;

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			m_nMaxElemInList	= -1;
			m_nMaxElemMul		= 0;

			for (int j = 0; j < nNumOfRuns; ++j)
			{		
				if (nNumOfRuns-1 == j)
				{
					nMaxResultSize		= (nRelByteCount - j*nMaxResultSize);
					nResultMaxRowsCount = (nMaxResultSize/ sizeof(T));
				}
				int nBufferOffset	= 0;
				int nNumOfElements	= 0;
				nCurrMaxMul			= 0;
				nLastItrElems		= 0;
				dwlOffset			= 0;

				for (int i = 0; i < nNumberOfBuffersToRead; ++i)
				{
					int nReadByteCount	= nReadSectorByteCount;
					int nRowsNum		= nSectorRowsNum;

					if (nNumberOfBuffersToRead-1 == i)
					{
						nReadByteCount	= nLastSectorByteCount;
						nRowsNum		= nLastSectorByteCount/sizeof(T);
					}
					psRelBuffer	= psResultBuffer + nBufferOffset;
							
					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nReadByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
					{		
						dwlOffset+= nReadByteCount;

						nResIndex	= 0;
						nIndex		= 0;

						// Take only the relevant elements.
						for (nIndex = 0; nIndex < nRowsNum; ++nIndex)
						{
							if (psRelBuffer[nIndex].GetKey() < m_nMaxElemInList)
							{
								continue;
							}

							if (psRelBuffer[nIndex].GetKey() == m_nMaxElemInList && nCurrMaxMul < m_nMaxElemMul)
							{
								nCurrMaxMul++;
								continue;
							}

							
							psRelBuffer[nResIndex] = psRelBuffer[nIndex];
							++nResIndex;
						}
					}

					//if (nLastItrElems == 0 && 0 == nResIndex)
					//{
					//	printf("%d \n", i);
					//}

					// 1.1 Sort the buffer of the relation
					QuickSort2(psResultBuffer, 0, nLastItrElems + nResIndex-1);

					nLastItrElems	= min(nLastItrElems + nResIndex, nResultMaxRowsCount);

					nBufferOffset	= nLastItrElems;

					psRelBuffer		= psResultBuffer + nBufferOffset;

				}

				// 4.1 Get the maximum number of this buffer and its multiplicity
		
				m_nMaxElemInList	= psResultBuffer[nLastItrElems-1].GetKey();

				m_nMaxElemMul = 0;
				for (int i = nLastItrElems-1; i > 0; --i)
				{
					if (psResultBuffer[i].GetKey() == m_nMaxElemInList)
					{
						m_nMaxElemMul++;
					}
					else
					{
						break;
					}
				}

				// 4.2 Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psResultBuffer, nMaxResultSize, dwlResultOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
				{							
					dwlResultOffset += nMaxResultSize;
				}
			}

				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);
		}
	}


	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psResultBuffer), 2*m_nBufferLimitedByteCount);

	return bRes;
}

// In this function we will have 3 buffers: input, output, working area. This is in order to decrease the computation time
// In this way we won't have to quick sort the input+output every iteration, but only sort the input, merge it with the output
// in the working area, and then copy it to the output again. In the first iteration we can read a whole buffer and sort it.
// In next work we will be able to change the input area dynamicaly. The working area must be always the same (maximum output size)
template <class T>
bool CSortTests::MergeSortKsV4(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
						int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum,
						double fReadBufferRatio)
{

	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList	= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul		= 1;

	// We devide the memory to read buffer, write buffer, working area buffer
	int nTotalNumOfSectors				= (m_nBufferLimitedByteCount/512);

	int nReadNumOfSectors				= static_cast<int>(floor(nTotalNumOfSectors/3.0f));

	int nWriteWorkNumOfSectors			= ((nTotalNumOfSectors - nReadNumOfSectors)/2);

	int nReadBufferLimitedByteCount		= (nReadNumOfSectors*512);

	int nWriteWorkBufferLimitedByteCount = (nWriteWorkNumOfSectors*512);

	if ((0!=(nReadBufferLimitedByteCount%512)) ||(0!=(nWriteWorkBufferLimitedByteCount%512))) 
	{
		printf("Handle it \n");
	}
	
	int nReadBuffersLimitedRowsNum				= (nReadBufferLimitedByteCount / sizeof(T));
	int nWriteWorkBuffersLimitedRowsNum			= (nWriteWorkBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuffersToRead					= static_cast<int>(floor((nRelByteCount-m_nBufferLimitedByteCount) /  static_cast<float>(nReadBufferLimitedByteCount)));
	int nLastReadBufferLimitedByteCount			= ((nRelByteCount-m_nBufferLimitedByteCount) - (nReadBufferLimitedByteCount*nNumberOfBuffersToRead));
//	int nLastReadBufferRowsNum					= (nLastReadBufferLimitedByteCount/sizoef(T));

	nNumberOfBuffersToRead++;
	if (nLastReadBufferLimitedByteCount > 0)
	{
		nNumberOfBuffersToRead++;
	}
	else
	{
		nLastReadBufferLimitedByteCount = nReadBufferLimitedByteCount;
	}

	unsigned int  nMaxResultSize				= nWriteWorkBuffersLimitedRowsNum;

	//We will need (relation_size/write_buffer_size) iteations
	int nNumOfRuns								= static_cast<int>(ceilf(nRelByteCount/static_cast<float>(nWriteWorkBufferLimitedByteCount)));
	
	// Allocate buffers

	T* psReadBuffer				= NULL;
	T* psWorkBuffer				= NULL;
	T* psWriteBuffer			= NULL;

	T* psBuffer			= NULL;

	if (AllocateVirtualBuffer((void**)(&psBuffer), m_nBufferLimitedByteCount ) == false)
	{
		return false;
	}

	// The buffer will be divided like this: (psWriteBuffer->psReadBuffer->psWorkBuffer)
	psReadBuffer		= psBuffer;
	psWriteBuffer		= psBuffer;
	psWorkBuffer		= psBuffer+(nWriteWorkBuffersLimitedRowsNum+nReadBuffersLimitedRowsNum);

	int nResIndex		= 0;
	int nIndex			= 0;
	int nCurrMaxMul		= 0;
	int nLastItrElems	= 0;

	int nLeftBytesToWrite = nRelByteCount;

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			m_nMaxElemInList	= -1;
			m_nMaxElemMul		= 0;

			for (int j = 0; j < nNumOfRuns; ++j)
			{			
				int nBufferOffset	= 0;
				int nNumOfElements	= 0;
				nCurrMaxMul			= 0;
				nLastItrElems		= 0;
				dwlOffset			= 0;

				for (int i = 0; i < nNumberOfBuffersToRead; ++i)
				{
					int nByteCount = nReadBufferLimitedByteCount;
					
					if (0 == i)
					{
						psReadBuffer	= psBuffer;
						// If this is the first iteration, write to the whole memory
						nByteCount		= m_nBufferLimitedByteCount;
					}
					else if (nNumberOfBuffersToRead-1 == i)
					{
						nByteCount = nLastReadBufferLimitedByteCount;
					}


					// 1. Read a buffer of the iteration

					if (true == psDiskRel->DoOverlappedIO(psReadBuffer, nByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
					{	
//						printf("Read %d bytes from offser %d \n", nByteCount, dwlOffset);
						dwlOffset+= nByteCount;

						nResIndex	= 0;
						nIndex		= 0;

						int nBufferLimtedRowsNum = (nByteCount/sizeof(T));

						if (j > 0)
						{
							// Take only the relevant elements.
							for (nIndex = 0; nIndex < nBufferLimtedRowsNum; ++nIndex)
							{
								if (psReadBuffer[nIndex].GetKey() < m_nMaxElemInList)
								{
									continue;
								}

								if (psReadBuffer[nIndex].GetKey() == m_nMaxElemInList && nCurrMaxMul < m_nMaxElemMul)
								{
									nCurrMaxMul++;
									continue;
								}
								
								psReadBuffer[nResIndex] = psReadBuffer[nIndex];
								++nResIndex;
							}
						}
						else
						{
							nResIndex = nBufferLimtedRowsNum;
						}
					}

					// 1.1 Sort the input buffer of the relation
					QuickSort2(psReadBuffer, 0, nResIndex-1);

					// 1.2 Merge the sorted input and sorted output into the working buffer - not for the first read buffer

					if (0 != i)
					{	
						int nLeftElements	= 0;
						int nWorkIndex		= 0;
						int nReadIndex		= 0;
						int nWriteIndex		= 0;

						while((nWorkIndex < nWriteWorkBuffersLimitedRowsNum) && (nReadIndex<nResIndex) && (nWriteIndex<nLastItrElems))
						{
							if (psWriteBuffer[nWriteIndex] <= psReadBuffer[nReadIndex])
							{
								psWorkBuffer[nWorkIndex++] = psWriteBuffer[nWriteIndex];
								nWriteIndex++;
							}
							else
							{
								psWorkBuffer[nWorkIndex++] = psReadBuffer[nReadIndex];
								nReadIndex++;
							}
						}

						if ((nWorkIndex < nWriteWorkBuffersLimitedRowsNum) && ((nReadIndex>=nResIndex) || (nWriteIndex>=nLastItrElems)))
						{					
							if ((nReadIndex<nResIndex) &&(nWriteIndex>=nLastItrElems))
							{
								// there are still elements in the read buffer
								nLeftElements = min(nWriteWorkBuffersLimitedRowsNum-nWorkIndex, nResIndex - nReadIndex);

								// Copy them to the work buffer
								memcpy(&(psWorkBuffer[nWorkIndex]), &(psReadBuffer[nReadIndex]), nLeftElements*sizeof(T));

								nWorkIndex+=nLeftElements;


							}
							else
							{
								// there are still elements in the write buffer
								nLeftElements = min(nWriteWorkBuffersLimitedRowsNum-nWorkIndex, nLastItrElems - nWriteIndex);

								// Copy them to the work buffer
								memcpy(&(psWorkBuffer[nWorkIndex]),  &(psWriteBuffer[nWriteIndex]), nLeftElements*sizeof(T));

								nWorkIndex+=nLeftElements;

							}
						}

						// 1.3 Copy the sorted result back to the output buffer
						memcpy(psWriteBuffer, psWorkBuffer, nWorkIndex*sizeof(T));

						nLastItrElems = min(nWorkIndex, nWriteWorkBuffersLimitedRowsNum);

					}
					else
					{
						// the first iteration
						nLastItrElems = min(nResIndex, nWriteWorkBuffersLimitedRowsNum);
					}

					psReadBuffer		= psBuffer + nWriteWorkBuffersLimitedRowsNum;

				}

				// 4.1 Get the maximum number of this buffer and its multiplicity
				
				m_nMaxElemInList	= psWriteBuffer[nLastItrElems-1].GetKey();
				m_nMaxElemMul		= 0;

				for (int i = nLastItrElems-1; i > 0; --i)
				{
					if (psWriteBuffer[i].GetKey() == m_nMaxElemInList)
					{
						m_nMaxElemMul++;
					}
					else
					{
						break;
					}
				}


				if (nLastItrElems < nWriteWorkBuffersLimitedRowsNum)
				{
					printf("CSortTests:: the buffer to write is not full \n");
				}

				// 4.2 Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psWriteBuffer, nWriteWorkBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
				{							
//					printf("Wrote %d bytes to offset %d \n",nWriteWorkBufferLimitedByteCount,  dwlResultOffset);

					dwlResultOffset += nWriteWorkBufferLimitedByteCount;

					nLeftBytesToWrite -= nWriteWorkBufferLimitedByteCount;

					if (nLeftBytesToWrite < nWriteWorkBufferLimitedByteCount)
					{
						// This is the last iteration and this is the left bytes to write
						nWriteWorkBufferLimitedByteCount	= nLeftBytesToWrite;
						nWriteWorkBuffersLimitedRowsNum		= (nWriteWorkBufferLimitedByteCount/sizeof(T));
					}

					
				}
			}

				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);
		}
	}


	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psBuffer), m_nBufferLimitedByteCount);

	return bRes;

}

// Description: Read the whole relation and keep the k smallest elements. Then write them and read the relation again and keep the 
//	next k smallest elements, and so on.
//	We will write the relation once and read the relation logk(M) times.

template <class T>
bool CSortTests::MergeSortKs(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
						int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
{
	rnReadSectorsNum		= 0;
	rnWrittenSectorsNum		= 0;

	bool bRes = true;

	// Initialization
	m_nMaxElemInList	= 0;
	// we keep the multiplicity of the maximum number
	m_nMaxElemMul		= 1;

	// the results buffer and the read buffer are the same size.
	int nBuffersLimitedRowsNum		= (m_nBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuffersToRead		= static_cast<int>(ceil(nRelByteCount /  static_cast<float>(m_nBufferLimitedByteCount)));

	unsigned int  nMaxResultSize	= nBuffersLimitedRowsNum;
	// We will need log(M) iterations - we will read the whole relation and take the smallest M.
	int nNumOfRuns					= nNumberOfBuffersToRead;//static_cast<int>(ceil(m_nBufferLimitedByteCount / static_cast<float>(nEveryBufferByteCount)));

	// Allocate buffers

	T* psRelBuffer				= NULL;
	T* psRelResultBuffer		= NULL;

	if (AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount ) == true)
	{
		if (AllocateVirtualBuffer((void**)(&psRelResultBuffer), m_nBufferLimitedByteCount ) != true)
		{
			DeAllocateVirtualBuffer((void**)(&psRelBuffer));
			return false;
		}
	}

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psDiskResult		= new Disk(wsSortedFileName, eLogical, eWrite);

	list<T> pSmallersList;

	if (psDiskRel->OpenDisk() && psDiskResult->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskResult->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlResultOffset	= 0;

			// Run over all iterations
				
			for (int i = 0; i < nNumOfRuns; ++ i)
			{
				dwlOffset		= 0;
				int nLastMax	= m_nMaxElemInList;
				int nLastMaxMul	= m_nMaxElemMul;
				int nFoundMax	= 0;

				// For the whole relation			
				for (int j = 0; j < nNumberOfBuffersToRead; ++j)
				{
					// TODO: handle reading a last not whole buffer
					// Read M bytes from the relation
					if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
					{		
						//rnReadSectorsNum+= m_nBufferLimitedByteCount/512;
						dwlOffset+= m_nBufferLimitedByteCount;
						// Update the smallest M list
						for (int k = 0; k < nBuffersLimitedRowsNum; ++k)
						{
							// Only if the element is bigger then the last maximum - inser it
							if ((nLastMax < psRelBuffer[k].GetKey()) || 
								((nLastMax == psRelBuffer[k].GetKey()) && (nFoundMax++ >= nLastMaxMul)))
							{
								
								InsertToResultList<T>(&(psRelBuffer[k]), &pSmallersList);
							}
						}
					}
					else
					{
						printf("Failed reading the relation \n");
					}
				}
				
				// Copy the data from the list to the buffer

				list<T>::iterator itr;
				int nIndex = 0;

				for(itr=pSmallersList.begin(); itr != pSmallersList.end(); ++itr)
				{
					psRelResultBuffer[nIndex] = *itr;
					++nIndex;
				}

				// Clear the list
				pSmallersList.clear();

				// Write the result list 
				if (true == psDiskResult->DoOverlappedIO(psRelResultBuffer, m_nBufferLimitedByteCount, dwlResultOffset, &rnWrittenSectorsNum))
				{							
					//rnWrittenSectorsNum+= m_nBufferLimitedByteCount/512;

					dwlResultOffset += m_nBufferLimitedByteCount;
				}
		
			}

		}
	}

	psDiskRel->CloseDisk();
	psDiskResult->CloseDisk();

	delete(psDiskRel);
	delete(psDiskResult);

	DeAllocateVirtualBuffer((void**)(&psRelBuffer));

	DeAllocateVirtualBuffer((void**)(&psRelResultBuffer));





	// Run over all iterations
		// For the whole relation
			// Read the M bytes from the relation
				// Update the smallest M list
		// Write the list. 
		// Keep the bigger number of this list.
			// TODO: handle the case in which the bigger number in the list appears more than once, but not the whole copies are in
			//			the written list - so the next time we will need to write the other copies.


	return bRes;
}

template <class T>
bool CSortTests::InsertToResultList(T* pNewElem, list<T>* pSmallersList)
{
	bool bRes = false;

	// the results buffer and the read buffer are the same size.
	int nBuffersLimitedRowsNum		= (m_nBufferLimitedByteCount / sizeof(T));

	unsigned int  uMaxResultSize	= nBuffersLimitedRowsNum;

	if (pNewElem->GetKey() > m_nMaxElemInList)
	{
		if (pSmallersList->size() < uMaxResultSize)
		{
			// If we have room in the list - 
			// Insret it to the end of the list and declare a new maximum
			pSmallersList->push_back(*pNewElem);
			m_nMaxElemInList = pNewElem->GetKey();
			// we keep the multiplicity of the maximum number
			m_nMaxElemMul  = 1;
			bRes = true;
		}
		else
		{
			// If the list is full - 
			// don't enter the list
		}
	}
	else if (pNewElem->GetKey() == m_nMaxElemInList)
	{
		// The new element key is as the last element in the list (maximum)
		if (pSmallersList->size() < uMaxResultSize)
		{
			// Inset to the end of the list and declare a new maximum
			pSmallersList->push_back(*pNewElem);
			
			// Increment the multiplicity of the maximum key in the list
			m_nMaxElemMul++;

			bRes = true;
		}
		else
		{
			// Don't enter the list
		}

	}
	else
	{
		// the new element has to be insert in the middle of the list
		list<T>::iterator itr;
		list<T>::iterator itr1;
		itr1 = (pSmallersList->end())--;

		for(itr=pSmallersList->begin(); itr != pSmallersList->end(); ++itr)
		{
			// Go over the whole list and find its place
			// TODO: maybe change it to a binary search
			if (*pNewElem <= *itr)
			{
				if (pSmallersList->size() < uMaxResultSize)
				{
					// The list is not full
					pSmallersList->insert(itr, *pNewElem);
					bRes = true;
				}
				else
				{
					// The list is full

					// Remove the last element and insert the new one in the proper place
					itr = pSmallersList->insert(itr,*pNewElem);
					//itr1 = pSmallersList->erase(itr1);
					pSmallersList->pop_back();

					bRes = true;

					if (m_nMaxElemMul > 1)
					{
						// The maximum number is still the same
						// decrease the multiplicity
						m_nMaxElemMul--;

					}
					else
					{
					
						// We have to find a new maximum
						// Check what is the multiplicity of the new max element
						itr1 = --pSmallersList->end();
						m_nMaxElemInList = (*itr1).GetKey();
						m_nMaxElemMul = 0;

						// Go over the list (in reverse) and calc the multiplicity
						while((*itr1).GetKey() == m_nMaxElemInList)
						{
							--itr1;
							m_nMaxElemMul++; 
						}
					}
				}
				// we inserted the new element - we can stop
				break;

			}

		}
	}

	return bRes;
}

bool CSortTests::AllocateVirtualBuffer(void** pBuffer, int nBytesCount)
{
	bool bIsReady = false;
	// Allocate a buffer
	VirtualFree(*pBuffer, 0, MEM_RELEASE);
	if (!(*pBuffer = VirtualAlloc(NULL, nBytesCount,  MEM_COMMIT, PAGE_READWRITE/*|PAGE_NOCACHE*/)))
	{
		cout << "*** Could not allocate buffer to prepare disk." << endl;
		bIsReady =  false;
	}
	else
	{
		VirtualLock((void**)(pBuffer), nBytesCount);
		bIsReady = true;
	}
	return bIsReady;
}

bool CSortTests::DeAllocateVirtualBuffer(void** pBuffer, int nBytesCount)
{
	VirtualFree(*pBuffer, 0, MEM_RELEASE);
	VirtualUnlock((void**)(pBuffer), nBytesCount);
	return true;
}

void CSortTests::SortingUnitTest(int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, int nRel2ByteCount, wstring wsRel1FileNameFlash,
								 wstring wsRel2FileNameFlash)
{

	//int a[999];
	//int b[999];

	//for (int i = 0; i < 999; ++i)
	//{
	//	a[i] = rand();
	//	b[i] = a[i];
	//}

	//a[0] = 5;
	//a[15] = 0;

	//b[0] = 5;
	//b[15] = 0;


	//DWORD dwStart = timeGetTime();
	//QuickSort2<int>(a, 0, 998);
	//DWORD dwEnd = timeGetTime();
	//float fTimeMsec = CalcTime(dwStart, dwEnd);
	//printf("new sort took %f [msec] \n", fTimeMsec);

	//dwStart = timeGetTime();
	//QuickSort<int>(b,0,999);
	//dwEnd = timeGetTime();
	//fTimeMsec = CalcTime(dwStart, dwEnd);
	//printf("old sort took %f [msec] \n", fTimeMsec);



	//for (int i = 0; i < 1000; ++i)
	//{
	//	if (a[i]!=b[i])
	//	{
	//		printf("%d - %d \n", a[i], b[i]);
	//	}
	//	if (i%100 == 0)
	//	{
	//		printf("********************************\n");
	//	}
	//}

	//return;
	
	wstring wsFinalFileName;
	float fTime				= 0;
	float fIoTime			= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;


#ifdef __ITERATIONS_SORT__

	int nTotalNumOfIterations = static_cast<int>(ceil((log(static_cast<float>(nRel2ByteCount/m_nBufferLimitedByteCount))/log(2.0f))));
	nTotalNumOfIterations++;
	SortInIterations<SRel2>(0, nTotalNumOfIterations, wsRel2FileNameFlash, L"\\\\.\\C:\\rel2Sort", nRel2RowsNum,
		nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum, wsFinalFileName);

#endif // __ITERATIONS_SORT__
#ifdef __SORT_IN_BLOCKS__
	wsFinalFileName= L"\\\\.\\C:\\rel2Sort";
	SortInBlocks<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1");
	wsFinalFileName.append(L"1");
#endif // __SORT_IN_BLOCKS__
#ifdef __SORT_MERGE_K__
			wsFinalFileName= L"\\\\.\\C:\\rel2Sort";
//			MergeSortKs<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//			MergeSortKsV1<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//		MergeSortKsV2<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum);
//		MergeSortKsV3<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, 0.6);
//		MergeSortKsV4<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum, 1.0);

		m_nBufferLimitedByteCount/=2;
		MergeSortKsV5<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, nRel2ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, 0.6);
		m_nBufferLimitedByteCount*=2;

#endif // __SORT_MERGE_K__
#ifdef __RADIX_SORT__
			wsFinalFileName= L"\\\\.\\C:\\rel2Sort";
			RadixSortV2<SRel2>(wsRel2FileNameFlash, wsFinalFileName, nRel2RowsNum, 1.0f, nRel2ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1");
#endif // __RADIX_SORT__

	// Check the results

	SRel2* psRel2Buffer = NULL;
	if (AllocateVirtualBuffer((void**)(&psRel2Buffer), nRel2ByteCount) != true)
	{
		printf("ERROR: cannot allocate results buffer \n");
	}


	Disk* psFlashDiskRel1	= new Disk(wsFinalFileName, eLogical, eRead);

	DWORDLONG dwlResOffset = 0;
	int nResIndex = 0;

	
	if ( psFlashDiskRel1->OpenDisk())
	{
		if (psFlashDiskRel1->CreateOverlappedEvent())
		{
			DWORD dwBytesRead			= 0;
			DWORDLONG dwlPrepareOffset	= 0;
			int nResultCounter			= 0;
			int nReadSectors			= 0;

			DWORD dwStart = timeGetTime();


			if (true == psFlashDiskRel1->DoOverlappedIO(psRel2Buffer, nRel2ByteCount, dwlResOffset))
			{
				for (int i = 1; i < nRel2RowsNum; ++i)
				{
					if (psRel2Buffer[i-1] > psRel2Buffer[i] && (0!= psRel2Buffer[i].GetKey()))
					{
						printf("not sorted [%d] = %d, [%d] = %d\n", i-1, psRel2Buffer[i-1].GetKey(), i ,  psRel2Buffer[i].GetKey());
					}
//					printf("%d, \n", psRel2Buffer[i-1].GetKey());
				}


				//printf("%d, \n", psRel2Buffer[nRel2RowsNum-1].GetKey());
			}
		}
	}
	psFlashDiskRel1->CloseDisk();
	delete(psFlashDiskRel1);
	DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nRel2ByteCount);


	return;
	// Perform another test:
	// This test compares between QuickSort and SortIterations

	// Check the correctnece of the sorting algs

	SRel1* psOriginalBuffer;

	bool bRes = AllocateVirtualBuffer((void**)(&psOriginalBuffer), nRel1ByteCount) ;

	for (int i = nRel1RowsNum-1; i >= 0; i--)
	{
		psOriginalBuffer[i].m_nc = i+1;
	}

	psOriginalBuffer[nRel1RowsNum-1].m_nc = 7;
	psOriginalBuffer[nRel1RowsNum/2].m_nc = 327;

	wstring sOriginal	= L"\\\\.\\C:\\Temp";
	wstring sSorted		= L"\\\\.\\C:\\TempSort";


	// Write the relation to the disk
	Disk* psDiskRel1 = new Disk(wsRel1FileNameFlash , eLogical, eWrite);
	if (psDiskRel1->OpenDisk() == true)
	{
		if( psDiskRel1->CreateOverlappedEvent())
		{
			if (psDiskRel1->DoOverlappedIO(psOriginalBuffer, nRel1ByteCount))
			{
				cout << "Wrote relation 1 to hard-disk" << endl;
			}
			else
			{
				cout << "Failed in writing relation 1 from hard-disk" << endl;
			}

			psDiskRel1->CloseDisk();		
		}
	}
	else
	{
		cout << "failed opening the disk" << endl;
	}
	delete psDiskRel1;

	QuickSort<SRel1>(psOriginalBuffer, 0, nRel1RowsNum);


	float	fTimeSec		= 0;
	nReadSectorsNum			= 0;
	nWrittenSectorsNum		= 0;



#ifdef __ITERATIONS_SORT__
	// Read it again and sort it
	nTotalNumOfIterations = static_cast<int>(ceil((log(static_cast<float>(nRel1ByteCount/m_nBufferLimitedByteCount))/log(2.0f))));
	nTotalNumOfIterations++;
	SortInIterations<SRel1>(0, nTotalNumOfIterations, wsRel1FileNameFlash, sSorted, nRel1RowsNum,
		nRel1ByteCount,fTimeSec,nReadSectorsNum, nWrittenSectorsNum, sSorted);

#endif //__ITERATIONS_SORT__
#ifdef __SORT_IN_BLOCKS__
	SortInBlocks<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, nRel1ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1");
	sSorted.append(L"1");

#endif // __SORT_IN_BLOCKS__
#ifdef __SORT_MERGE_K__
	//		MergeSortKs<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, nRel1ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
	//			MergeSortKsV1<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, nRel1ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
	//	MergeSortKsV2<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, nRel1ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
		MergeSortKsV4<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, nRel1ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum, 1.0);

#endif // __SORT_MERGE_K__
#ifdef __RADIX_SORT__
			RadixSortV2<SRel1>(wsRel1FileNameFlash, sSorted, nRel1RowsNum, 1.0f, nRel1ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1");

#endif // __RADIX_SORT__

		
	bRes = AllocateVirtualBuffer((void**)(&m_psRel1Buffer), nRel1ByteCount) ;
	
	psDiskRel1 = new Disk(sSorted , eLogical, eRead);
	
	if (psDiskRel1->OpenDisk() == true)
	{
		if( psDiskRel1->CreateOverlappedEvent())
		{
			if (psDiskRel1->DoOverlappedIO(m_psRel1Buffer, nRel1ByteCount))
			{
				cout << "Read relation 1 from hard-disk" << endl;
			}
			else
			{
				cout << "Failed in reading relation 1 from hard-disk" << endl;
			}
			psDiskRel1->CloseDisk();		
		}
	}
	else
	{
		cout << "failed opening the disk" << endl;
	}
	delete psDiskRel1;

	int nIndex = 0;

	for (int i = 0; i < nRel1RowsNum; ++i)
	{
		/*printf("%d \n", m_psRel1Buffer[i].GetKey());*/
		
		if (0 == m_psRel1Buffer[i].GetKey())
		{
			continue;
		}

		if (m_psRel1Buffer[i].GetKey() != psOriginalBuffer[nIndex].GetKey())
		{
			printf("***problem:*** %d. %d-%d \n", i, m_psRel1Buffer[i].GetKey() , psOriginalBuffer[nIndex].GetKey());
		}
		nIndex++;
	}

	DeAllocateVirtualBuffer((void**)m_psRel1Buffer, nRel1ByteCount);

	DeAllocateVirtualBuffer((void**)psOriginalBuffer, nRel1ByteCount);

	return;

}


void CSortTests::SortingUnitTest1(int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, int nRel2ByteCount, wstring wsRel1FileNameFlash,
								 wstring wsRel2FileNameFlash)
{

	wstring wsFinalFileName;
	float fTime				= 0;
	float fIoTime			= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;


	// Perform another test:
	// This test compares between QuickSort and almostSorted

	// Check the correctnece of the sorting algs

	SRel2* psOriginalBuffer;

	bool bRes = AllocateVirtualBuffer((void**)(&psOriginalBuffer), nRel2ByteCount) ;

	//for (int i = 0; i < nRel2RowsNum; i++)
	//{
	//	psOriginalBuffer[i].m_nc = i+1;
	//}

	//SRel2 sTemp				= psOriginalBuffer[5];
	//psOriginalBuffer[5]		= psOriginalBuffer[641];
	//psOriginalBuffer[641]	= sTemp;


	//sTemp						= psOriginalBuffer[300];
	//psOriginalBuffer[300]		= psOriginalBuffer[21];
	//psOriginalBuffer[21]		= sTemp;


	//sTemp						= psOriginalBuffer[1000];
	//psOriginalBuffer[1000]		= psOriginalBuffer[3000];
	//psOriginalBuffer[3000]		= sTemp;


	wstring sOriginal	= L"\\\\.\\C:\\TempFile";
	wstring sSorted		= L"\\\\.\\C:\\TempSorteFile";
	wstring sFinalSorted = L"\\\\.\\C:\\TempSorteFile1";


	// Write the relation to the disk
	Disk* psDiskRel2 = new Disk(wsRel2FileNameFlash , eLogical, eRead);
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(psOriginalBuffer, nRel2ByteCount))
			{
				cout << "Read relation 2 from hard-disk" << endl;
			}
			else
			{
				cout << "Failed in writing relation 1 from hard-disk" << endl;
			}

			psDiskRel2->CloseDisk();		
		}
	}
	else
	{
		cout << "failed opening the disk" << endl;
	}
	delete psDiskRel2;

	QuickSort<SRel2>(psOriginalBuffer, 0, nRel2RowsNum);


	float	fTimeSec		= 0;
	nReadSectorsNum			= 0;
	nWrittenSectorsNum		= 0;



//	AlmostSortedMerge<SRel2>(sOriginal, sSorted, nRel2RowsNum, nRel2ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum);

//	MergeSortKsV3<SRel2>(wsRel2FileNameFlash, sSorted, nRel2RowsNum, nRel2ByteCount,fTime,&fIoTime, nReadSectorsNum, nWrittenSectorsNum, 1.0, false);
	SortInBlocksV1<SRel2>(wsRel2FileNameFlash, sSorted, nRel2RowsNum, nRel2ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1", 0.4);
//	SortInBlocks<SRel2>(wsRel2FileNameFlash, sSorted, nRel2RowsNum, nRel2ByteCount,fTime, &fIoTime, nReadSectorsNum, nWrittenSectorsNum, L"1");



		
	bRes = AllocateVirtualBuffer((void**)(&m_psRel2Buffer), nRel2ByteCount) ;
	
	psDiskRel2 = new Disk(sFinalSorted/*sSorted*/ , eLogical, eRead);
	
	if (psDiskRel2->OpenDisk() == true)
	{
		if( psDiskRel2->CreateOverlappedEvent())
		{
			if (psDiskRel2->DoOverlappedIO(m_psRel2Buffer, nRel2ByteCount))
			{
				cout << "Read relation 2 from hard-disk" << endl;
			}
			else
			{
				cout << "Failed in reading relation 2 from hard-disk" << endl;
			}
			psDiskRel2->CloseDisk();		
		}
	}
	else
	{
		cout << "failed opening the disk" << endl;
	}
	delete psDiskRel2;

	int nIndex = 0;

	for (int i = 0; i < nRel2RowsNum; ++i)
	{
		/*printf("%d \n", m_psRel1Buffer[i].GetKey());*/
		
		if (0 == m_psRel2Buffer[i].GetKey())
		{
			continue;
		}

		if (m_psRel2Buffer[i].GetKey() != psOriginalBuffer[nIndex].GetKey())
		{
			printf("***problem:*** %d. %d-%d \n", i, m_psRel2Buffer[i].GetKey() , psOriginalBuffer[nIndex].GetKey());
			break;
		}
		nIndex++;
	}

	DeAllocateVirtualBuffer((void**)m_psRel2Buffer, nRel2ByteCount);

	DeAllocateVirtualBuffer((void**)psOriginalBuffer, nRel2ByteCount);

	return;

}

void CSortTests::SetBufferLimitedByteCount(int nBufferLimitedByteCount)
{
	m_nBufferLimitedByteCount = nBufferLimitedByteCount;
}

void CSortTests::Test1(ostream& fResults, int nIteration, int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, int nRel2ByteCount,
					  int nSmallRelByteCount, int nBigRelByteCount, wstring wsRel2FileName)
{

	wstring wsFinalFileName;

	float fTimeMsec			= 0;
	float fIoTimeMsec		= 0;
	float fTime				= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;

	fResults << m_nBufferLimitedByteCount << "\t\t\t";
	fResults << nSmallRelByteCount << "\t\t" << nBigRelByteCount << "\t\t";

//	// Sort in blocks
//
//	nReadSectorsNum		= 0;
//	nWrittenSectorsNum	= 0;
//	fTimeMsec			= 0;
//	fIoTimeMsec			= 0;
//
//	printf("Merge Sort\n");
//	wstring wsAppend = L"2";
//#ifdef __FLASH__
//	SortInBlocks<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_2", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
//#else
//	SortInBlocks<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_2", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
//#endif // __FLASH__
//
//	printf("IO time is: %f [msec] \n", fIoTimeMsec);
//	printf("Total program time is: %f [msec]\n",  fTimeMsec);
//
//	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//	fResults << endl;
//
//
//	nReadSectorsNum		= 0;
//	nWrittenSectorsNum	= 0;
//	fTimeMsec			= 0;
//	fIoTimeMsec			= 0;
//
//	printf("Regular smaller\n");
//#ifdef __FLASH__
//	MergeSortKsV3<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, 0.6, false);
//
//#else
//	MergeSortKsV3<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, 1.0, false);
//#endif //  __FLASH__
//
//	printf("Erase Units smaller\n");
//	printf("IO time is: %f [msec] \n", fIoTimeMsec);
//	printf("Total program time is: %f [msec]\n",  fTimeMsec);
//
//	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//	fResults << endl;

	//fResults << endl;

	//Sleep(1000);

	// Merge Sort
	double a[9] = {0.3, 0.4, 0.5, 0.6, 0.7, 0.8,0.9, 1.0, 1.1};
	for (int i = 0; i < 9; ++i)
	{
		double fStep = a[i];
		printf("**************Ratio = %lf ***************\n", fStep);
		fResults << endl << "\t\t " << fStep << "\t";
		nReadSectorsNum		= 0;
		nWrittenSectorsNum	= 0;
		fTimeMsec			= 0;
		fIoTimeMsec			= 0;

#ifdef __FLASH__
//	MergeSortKs<SRel2>(m_wsRel2FileNameFlash, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//	MergeSortKsV1<SRel2>(m_wsRel2FileNameFlash, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//	MergeSortKsV2<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
	MergeSortKsV3<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, fStep, false);
//	MergeSortKsV5<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, fStep);
//	SortInBlocksV1<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_1", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, L"1", fStep);


#else
//	MergeSortKs<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//	MergeSortKsV1<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
//	MergeSortKsV2<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec,&fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//	MergeSortKsV3<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, fStep);
//	MergeSortKsV5<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_3", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, fStep);
	SortInBlocksV1<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_1", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, L"1", fStep);



#endif //  __FLASH__

	printf("IO time is: %f [msec] \n", fIoTimeMsec);
	printf("Total program time is: %f [msec]\n",  fTimeMsec);

	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

	//fResults << endl;

	Sleep(1000);
	

	}

//	// Radix sort
//	nReadSectorsNum		= 0;
//	nWrittenSectorsNum	= 0;
//	fTimeMsec			= 0;
//	fIoTimeMsec			= 0;
//
//
////	wsAppend = L"4";
////#ifdef __FLASH__
//////	RadixSortV1<SRel2>(wsRel2FileName,  L"\\\\.\\E:\\rel2Sort_4", nRel2RowsNum, 1.0f, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
////	RadixSortV2<SRel2>(wsRel2FileName,  L"\\\\.\\E:\\rel2Sort_4", nRel2RowsNum, 1.0f, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
////
////#else
//////	RadixSortV1<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_4", nRel2RowsNum, 1.0f, nRel2ByteCount,fTime,nReadSectorsNum, nWrittenSectorsNum);
////	RadixSortV2<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_4", nRel2RowsNum, 1.0f, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
////
////#endif // __FLASH__
////
////	printf("IO time is: %f [msec] \n", fIoTimeMsec);
////	printf("Total program time is: %f [msec]\n",  fTimeMsec);
////
////	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//
//#ifdef __SORTED_PIECES__
//	// merge sort of relation that conatains some sorted parts
//	nReadSectorsNum		= 0;
//	nWrittenSectorsNum	= 0;
//	fTimeMsec			= 0;
//	fIoTimeMsec			= 0;
//
//#ifdef __FLASH__
//
//	PiecewiseSortedMerge<SRel2>(wsRel2FileName,  L"\\\\.\\E:\\rel2Sort_5", nRel2RowsNum,
//					nRel2ByteCount, fTimeMsec,  &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//
//#else
//	PiecewiseSortedMerge<SRel2>(wsRel2FileName,  L"\\\\.\\C:\\rel2Sort_5", nRel2RowsNum,
//					nRel2ByteCount, fTimeMsec,  &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//
//
//#endif __FLASH__
//
//
//
//	printf("IO time is: %f [msec] \n", fIoTimeMsec);
//	printf("Total program time is: %f [msec]\n",  fTimeMsec);
//
//	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);
//
//#endif // __SORTED_PIECES__

	fResults << endl;
}


void CSortTests::Test(ostream& fResults, int nIteration, int nRel1RowsNum, int nRel2RowsNum, int nRel1ByteCount, int nRel2ByteCount,
					  int nSmallRelByteCount, int nBigRelByteCount, wstring wsRel2FileName, wstring wAppend)
{
	wstring wsFinalFileName;

	float fTimeMsec			= 0;
	float fIoTimeMsec		= 0;
	float fTime				= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;

	fResults << m_nBufferLimitedByteCount << "\t\t\t";
	fResults << nSmallRelByteCount << "\t\t" << nBigRelByteCount << "\t\t";

	// Sort in blocks

	nReadSectorsNum		= 0;
	nWrittenSectorsNum	= 0;
	fTimeMsec			= 0;
	fIoTimeMsec			= 0;

	wstring wsAppend = L"2";
#ifdef __FLASH__
	SortInBlocks<SRel2>(wsRel2FileName, L"\\\\.\\E:\\rel2Sort_2", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
#else
	SortInBlocks<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_2", nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);
#endif // __FLASH__

	printf("IO time is: %f [msec] \n", fIoTimeMsec);
	printf("Total program time is: %f [msec]\n",  fTimeMsec);

	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

	nReadSectorsNum		= 0;
	nWrittenSectorsNum	= 0;
	fTimeMsec			= 0;
	fIoTimeMsec			= 0;

	printf("false\n");
#ifdef __FLASH__
	wstring wsResFileName = L"\\\\.\\E:\\rel2Sort_3";
	wsResFileName.append(wAppend);
	MergeSortKsV3<SRel2>(wsRel2FileName,wsResFileName , nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, 0.5, false);
#else
	wstring wsResFileName = L"\\\\.\\C:\\rel2Sort_3";
	wsResFileName.append(wAppend);

	MergeSortKsV3<SRel2>(wsRel2FileName, wsResFileName, nRel2RowsNum, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, 0.5, false);
#endif //  __FLASH__

	printf("IO time is: %f [msec] \n", fIoTimeMsec);
	printf("Total program time is: %f [msec]\n",  fTimeMsec);

	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

	// Radix sort
	nReadSectorsNum		= 0;
	nWrittenSectorsNum	= 0;
	fTimeMsec			= 0;
	fIoTimeMsec			= 0;
	float fDencity		= 1.0;

//#ifdef __SORTED_PIECES__
//	fDencity = 0.2f;
//#endif // __SORTED_PIECES__


	wsAppend = L"4";
#ifdef __FLASH__
	RadixSortV2<SRel2>(wsRel2FileName,  L"\\\\.\\E:\\rel2Sort_4", nRel2RowsNum, fDencity, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);

#else
	RadixSortV2<SRel2>(wsRel2FileName, L"\\\\.\\C:\\rel2Sort_4", nRel2RowsNum, fDencity, nRel2ByteCount,fTimeMsec, &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum, wsAppend);

#endif // __FLASH__

	printf("IO time is: %f [msec] \n", fIoTimeMsec);
	printf("Total program time is: %f [msec]\n",  fTimeMsec);

	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);


#ifdef __SORTED_PIECES__
	// merge sort of relation that conatains some sorted parts
	nReadSectorsNum		= 0;
	nWrittenSectorsNum	= 0;
	fTimeMsec			= 0;
	fIoTimeMsec			= 0;

#ifdef __FLASH__

	PiecewiseSortedMerge<SRel2>(wsRel2FileName,  L"\\\\.\\E:\\rel2Sort_5", nRel2RowsNum,
					nRel2ByteCount, fTimeMsec,  &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

#else
	PiecewiseSortedMerge<SRel2>(wsRel2FileName,  L"\\\\.\\C:\\rel2Sort_5", nRel2RowsNum,
					nRel2ByteCount, fTimeMsec,  &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);


#endif __FLASH__



	printf("IO time is: %f [msec] \n", fIoTimeMsec);
	printf("Total program time is: %f [msec]\n",  fTimeMsec);

	PrintResults(fResults, fTimeMsec, fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

#endif // __SORTED_PIECES__

	fResults << endl;
}

void CSortTests::PrintResults(ostream& fResults, float fTimeSec, int nReadSectorsNum)
{
	fResults << fTimeSec / 1000.0f<< "\t" ;/*<< "-";*/
	fResults << nReadSectorsNum << "\t\t";
}

void CSortTests::PrintResults(ostream& fResults, float fTimeMsec, float fIoTimeMsec, int nReadSectorsNum, int nWrittenSectorsNum)
{
	fResults << fTimeMsec / 1000.0f<< "\t" ;/*<< "-";*/
	fResults << fIoTimeMsec / 1000.0f<< "\t" ;/*<< "-";*/

	fResults << nReadSectorsNum << "\t\t";
	fResults << nWrittenSectorsNum << "\t\t\t";

	Sleep(1000);

}


float CSortTests::CalcTime(DWORD dwStart, DWORD dwStop)
{
	DWORD dwDiffMsec = (dwStop - dwStart);
	float fDiffSec = static_cast<float>(dwDiffMsec)  / (1000.0f);
#ifdef __FLASH_DEBUG__
	cout << "It took " << dwDiffMsec << " msec " << fDiffSec << " seconds - " << fDiffSec / 60.0f << " minutes" <<endl;
#endif __FLASH_DEBUG__
	//return fDiffSec;
	return static_cast<float>(dwDiffMsec);
}


template <class T>
bool CSortTests::RadixSortV1(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum, float fColumDensity,
								int nRelByteCount, float& rfTimeSec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
{
	// 1. Initialization
	int nEveryBufferRowsCount	= (m_nBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuckets		= static_cast<int>(ceil(nRelRowsNum/(0.8f*nEveryBufferRowsCount)));

	int nEveryBucketByteCount	= static_cast<int>(floor(m_nBufferLimitedByteCount / static_cast<float>(nNumberOfBuckets)));

	int nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

	int nLastBucketByteCount	= nEveryBucketByteCount;

	int nLastBucketRowsCount	= nEveryBucketRowsCount;

	// Handle a case in which this number is not whole
	int nNumberOfBlocksInRel	= (nRelByteCount/m_nBufferLimitedByteCount);

	// The maximum read/write buffer is 65024 (127 sectors)
	//if (nEveryBucketByteCount > 51712)
	//{
	//	nEveryBucketByteCount	= 51712;

	//	nNumberOfBuckets		= static_cast<int>(ceil(static_cast<float>(nRelRowsNum)/nEveryBucketByteCount));

	//	nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

	//	nLastBucketRowsCount	= nEveryBucketRowsCount;

	//}

	// The buckets must be sectors align

	if (0 != (nEveryBucketByteCount % 512) || (0 != (nRelRowsNum % nEveryBucketByteCount)))
	{
		int nEveryBucketSecorsCount = (nEveryBucketByteCount / 512);

		if (0 == nEveryBucketSecorsCount)
		{

			nEveryBucketByteCount	= 512;

			nNumberOfBuckets		= static_cast<int>(ceil( static_cast<float>(m_nBufferLimitedByteCount)/nEveryBucketByteCount));

			nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

			nLastBucketByteCount	= nEveryBucketByteCount;

			nLastBucketRowsCount	= nEveryBucketRowsCount;
			
			nEveryBucketSecorsCount = (nEveryBucketByteCount / 512);

		}

		nEveryBucketByteCount	= (nEveryBucketSecorsCount * 512);

		nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

		nLastBucketByteCount	= (m_nBufferLimitedByteCount - (nEveryBucketByteCount*(nNumberOfBuckets-1)));

		nLastBucketRowsCount	= (nLastBucketByteCount/sizeof(T));

	}

	// 1.1 Create the buckets list
	T** psRelBucketsBuffer 		= (T**)malloc(nNumberOfBuckets*sizeof(T*));
	T* psRelBuffer				= NULL;

	// TODO: handle when nEveryBucketByteCount is less then a sector
	if (nEveryBucketByteCount < 512 || 0 != (nEveryBucketByteCount % 512))
	{
		printf("RadixSortV1 - Problem - not 512 aligned\n");
		return false;
	}

	bool bRes = true;
	int i = 0;
	int j = 0;

	// 1.2 Allocate virtual memory for the buckets
	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		int nByteCount = nEveryBucketByteCount; 
		// If this is the last bucket
		if ( i == nNumberOfBuckets-1)
		{
			nByteCount = nLastBucketByteCount;
		}
		if (AllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[i])), nByteCount ) != true)
		{
			bRes = false;
			break;
		}
	}

	if (false == bRes)
	{
		// Deallocate the previous buffers that were successfully been allocated
		// TODO: check if we have deallocate until i or until i-1
		for (j = 0; j < i; ++j)
		{
			DeAllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[j])));
		}

		return false;
	}

	// 1.3 Allocate virtual memory for the relation
	if (AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount ) != true)
	{
		bRes = false;		
	}


	// This disk is for reading the relation
	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	
	// The results file name will be wsSortedFileName + 1
	wsRelFileName.append(L"1");

	// This disk is for writing the intermediate results (the buckets)
	Disk* psDiskSortedRel	= new Disk(wsRelFileName, eLogical, eWrite);

	// Allocating the buckets map
	SBucketsMapping** asBucketsMap = static_cast<SBucketsMapping**>(malloc(nNumberOfBuckets*sizeof(SBucketsMapping*)));

	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		asBucketsMap[i] = new SBucketsMapping();
	}


	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlSoretdOffset	= 0;
			int nSortedBufferIndex		= 0;

			// Calculate the key range and the number of keys in every bucket
			int nKeysRange					= static_cast<int>(fColumDensity*nRelRowsNum);
			int nNumberOfKeysInABucket		= static_cast<int>(floor( static_cast<float>(nKeysRange)/nNumberOfBuckets));
			int nNumberOfKeysInLastBucket	= nNumberOfKeysInABucket;
			
			if (0 != (nKeysRange % nNumberOfBuckets))
			{
				nNumberOfKeysInLastBucket = (nKeysRange - nNumberOfKeysInABucket*(nNumberOfBuckets-1));
			}
			

			// Read the relation and divide to the buckets.
			
			for (i = 0; i < nNumberOfBlocksInRel; ++i)
			{
				int nBucketIndex = 0;
				if (true == psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum))
				{
					dwlOffset += m_nBufferLimitedByteCount;

					for (j = 0; j < nEveryBufferRowsCount; ++j)
					{
						int nKey = psRelBuffer[j].GetKey();
						nBucketIndex					=  static_cast<int>(floor( static_cast<float>(psRelBuffer[j].GetKey())/nNumberOfKeysInABucket));
						if (nBucketIndex >= nNumberOfBuckets)
						{
//							printf("this is to the last bucket = index: %d \n", nBucketIndex);
							nBucketIndex = nNumberOfBuckets-1;
						}

						int nIndex									= asBucketsMap[nBucketIndex]->m_nNumOfElems;
						*(psRelBucketsBuffer[nBucketIndex]+nIndex)	= psRelBuffer[j];
						asBucketsMap[nBucketIndex]->m_nNumOfElems++;

						// When a bucket is full - write it to disk.
						// TODO: try to write a whole block (32 sectors) - maybe it doesn't matter - the first one will cost more.

						int nCurrBucketIndex	= asBucketsMap[nBucketIndex]->m_nNumberOfParts;
						int nBucketLimit		= nEveryBucketRowsCount;

						if (nBucketIndex == nNumberOfBuckets -1)
						{
							// this is the last bucket so the number of elements in it is different
							nBucketLimit = nLastBucketRowsCount;

						}

						SBucketsMapping* pCurrBucket = asBucketsMap[nBucketIndex];

						while(NULL != pCurrBucket->m_pNextPart)
						{
							pCurrBucket = pCurrBucket->m_pNextPart;
						}

						if (nBucketLimit == asBucketsMap[nBucketIndex]->m_nNumOfElems)
						{
							if (true == psDiskSortedRel->DoOverlappedIO((psRelBucketsBuffer[nBucketIndex]), nBucketLimit*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum))
							{
								/*printf("Bucket = %d \n", nBucketIndex);*/
								//for (int k = 0; k < nBucketLimit; ++k)
								//{
								//	printf("%d, ", (psRelBucketsBuffer[nBucketIndex] + k)->GetKey());
								//}
								// Keep the offsets of the bucket part in a list.
								pCurrBucket->m_dwPartOffset		= dwlSoretdOffset;
								pCurrBucket->m_pNextPart		= new SBucketsMapping();
								pCurrBucket->m_nBytesCount		= nBucketLimit*sizeof(T);
								
								asBucketsMap[nBucketIndex]->m_nNumberOfParts++;
								asBucketsMap[nBucketIndex]->m_nNumOfElems	= 0;
								

								dwlSoretdOffset += nBucketLimit*sizeof(T);
							}
							else
							{
								printf("CSortTests:: failed in psDiskSortedRel->DoOverlappedIO\n");
							}

						}


					}
				}

			}

			// Finish writing the rest of the bucket
			// If there is not full bucket -> wrap it with 0
			for (j = 0; j < nNumberOfBuckets; ++j)
			{
				int nCurrBucketIndex		= asBucketsMap[j]->m_nNumberOfParts;
				int nNumOfElems				= (asBucketsMap[j])->m_nNumOfElems;
				if (nNumOfElems > 0)
				{
					int nSectorRowsCount		= (512/sizeof(T));
					int nNumOfSectors			= ((nNumOfElems / nSectorRowsCount) + 1);
					int nTotalElems				= (nSectorRowsCount * nNumOfSectors);
					int nRestOfElemsToSector	= (nTotalElems - nNumOfElems);
					

					T tempElem;

					for (i = nNumOfElems; i < nTotalElems; ++i)
					{
						*(psRelBucketsBuffer[j]+i) = tempElem;
					}

					SBucketsMapping* pCurrBucket = asBucketsMap[j];
					while(NULL != pCurrBucket->m_pNextPart)
					{
						pCurrBucket = pCurrBucket->m_pNextPart;
					}

					if (true == psDiskSortedRel->DoOverlappedIO((psRelBucketsBuffer[j]), nNumOfSectors*512, dwlSoretdOffset, &rnWrittenSectorsNum))
					{
						/*printf("Bucket = %d \n", j);*/
						// Keep the offsets of the bucket part in a list.
						pCurrBucket->m_dwPartOffset	= dwlSoretdOffset;
						pCurrBucket->m_nBytesCount	= nNumOfSectors*512;
						
						asBucketsMap[j]->m_nNumberOfParts++;
							
						dwlSoretdOffset += nNumOfSectors*512;
					}
					else
					{
						printf("CSortTests:: failed in psDiskSortedRel->DoOverlappedIO\n");
					}
				}
			}

		}

		psDiskRel->CloseDisk();
		psDiskSortedRel->CloseDisk();
	}

	// Closing the disks and deallocating the buffers

	delete(psDiskSortedRel);
	delete(psDiskRel);

	for (j = 0; j < nNumberOfBuckets; ++j)
	{
		DeAllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[j])));
	}



	// This disk is for reading the relation divided to buckets
	psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	
	// This disk is for writing the sorted relation
	 psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

	T* psSortedBuffer = NULL;

	if (AllocateVirtualBuffer((void**)(&psSortedBuffer), m_nBufferLimitedByteCount ) == true)
	{

		if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
		{
			if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
			{
				DWORDLONG dwlOffset			= 0;
				DWORDLONG dwlSoretdOffset	= 0;

				// For every bucket - read its parts from the disk
				for (i = 0; i < nNumberOfBuckets; ++i)
				{
					SBucketsMapping* pPart		= NULL;

					pPart = asBucketsMap[i];

					int nBucketRowsCount = 0;
					int nTotalRowsCount = 0;


					// Go to the end of the list and release the memory
					while ((NULL != pPart) && (0 != pPart->m_nBytesCount))
					{
						dwlOffset			= pPart->m_dwPartOffset;
						int nBytesCount		= pPart->m_nBytesCount;
						nBucketRowsCount	= (nBytesCount/sizeof(T));

						if (true != psDiskRel->DoOverlappedIO(&(psRelBuffer[nTotalRowsCount]), nBytesCount, dwlOffset, &rnReadSectorsNum))
						{
							printf("Problem in reading buckets\n");
							break;
						}
						
						nTotalRowsCount		+= nBucketRowsCount;

						pPart = pPart->m_pNextPart;
					}


					// Sort the whole bucket
					QuickSort<T>(psRelBuffer, 0, nTotalRowsCount);

					int nTotalBytesCount =  nTotalRowsCount*sizeof(T);

					// There might be some 0 to wrap the sector

					if (true != psDiskSortedRel->DoOverlappedIO(psRelBuffer, nTotalBytesCount, dwlSoretdOffset, &rnWrittenSectorsNum))
					{
						printf("Problem in reading buckets\n");
						break;
					}

					dwlSoretdOffset+= nTotalBytesCount;
				}


			}
			psDiskRel->CloseDisk();
			psDiskSortedRel->CloseDisk();
		}
		DeAllocateVirtualBuffer((void**)(&psSortedBuffer));

	}
			
	// Closing the disks and deallocating the buffers

	delete(psDiskSortedRel);
	delete(psDiskRel);

	DeAllocateVirtualBuffer((void**)(&psRelBuffer));

	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		SBucketsMapping* pPart		= NULL;
		SBucketsMapping* pNextPart	= NULL;

		pPart = asBucketsMap[i];

		// Go to the end of the list and release the memory
		while (pPart != NULL)
		{
			pNextPart = pPart->m_pNextPart;
			delete(pPart);
			pPart = pNextPart;
		}
		
		asBucketsMap[i] = NULL;
	}

	free(asBucketsMap);


	

	return bRes;
}


template <class T>
bool CSortTests::RadixSortV2(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum, float fColumDensity,
								int nRelByteCount, float& rfTimeSec, float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, wstring wsAppend)
{
	// 1. Initialization
	int nHalfBufferLimitedByteCount = (m_nBufferLimitedByteCount/2);

	int nEveryBufferRowsCount		= (nHalfBufferLimitedByteCount / sizeof(T));

	int nNumberOfBuckets			= static_cast<int>(ceil(nRelRowsNum/(0.8f*nEveryBufferRowsCount)));

	int nEveryBucketTotalByteCount	= static_cast<int>(floor(nRelRowsNum / static_cast<float>(nNumberOfBuckets)));

	int nEveryBucketByteCount		= static_cast<int>(floor(nHalfBufferLimitedByteCount / static_cast<float>(nNumberOfBuckets)));

	int nEveryBucketRowsCount		= (nEveryBucketByteCount/sizeof(T));

	int nLastBucketByteCount		= nEveryBucketByteCount;

	int nLastBucketRowsCount		= nEveryBucketRowsCount;

	// Handle a case in which this number is not whole
	int nNumberOfBlocksInRel		= (nRelByteCount/nHalfBufferLimitedByteCount);


	// The buckets must be sectors align

	int nEveryBucketSecorsCount = (nEveryBucketByteCount / 512);

	if (0 == nEveryBucketSecorsCount)
	{
		// the minimum size is 512 bytes

		nEveryBucketByteCount	= 512;

		nNumberOfBuckets		= static_cast<int>(ceil( static_cast<float>(nHalfBufferLimitedByteCount)/nEveryBucketByteCount));

		nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

		nLastBucketByteCount	= nEveryBucketByteCount;

		nLastBucketRowsCount	= nEveryBucketRowsCount;
		
		nEveryBucketSecorsCount = (nEveryBucketByteCount / 512);

	}
	else
	{
		// Every buffer will be aligned to 512
		nEveryBucketByteCount	= (nEveryBucketSecorsCount * 512);


		// We devide the total remainder of every buffer to the buffer size and get more buckets with the same size
		int nMoreBucketsNum		= ((nHalfBufferLimitedByteCount - (nEveryBucketByteCount*nNumberOfBuckets)) / nEveryBucketByteCount);

		nNumberOfBuckets		+= nMoreBucketsNum;

		int nRemainder			= (nHalfBufferLimitedByteCount - (nEveryBucketByteCount*nNumberOfBuckets));

		// Maybe we will still need a small last bucket
		if ( nRemainder > 0)
		{
			nNumberOfBuckets++;
			nLastBucketByteCount = nRemainder;

			if (0 != (nRemainder%512))
			{
				printf("Problem = 0 != (nRemaider%512) \n");
			}
		}
		else
		{
			nLastBucketByteCount = nEveryBucketByteCount;
		}

		nLastBucketRowsCount	= (nLastBucketByteCount/sizeof(T));

		nEveryBucketRowsCount	= (nEveryBucketByteCount/sizeof(T));

	}

	// 1.1 Create the buckets list
	T** psRelBucketsBuffer 		= (T**)malloc(nNumberOfBuckets*sizeof(T*));
	T* psRelBuffer				= NULL;

	// TODO: handle when nEveryBucketByteCount is less then a sector
	if (nEveryBucketByteCount < 512 || 0 != (nEveryBucketByteCount % 512))
	{
		printf("RadixSortV1 - Problem - not 512 aligned\n");
		return false;
	}

	bool bRes = true;
	int i = 0;
	int j = 0;

	// 1.2 Allocate virtual memory for the buckets
	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		int nByteCount = nEveryBucketByteCount; 
		// If this is the last bucket
		if ( i == nNumberOfBuckets-1)
		{
			nByteCount = nLastBucketByteCount;
		}
		if (AllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[i])), nByteCount ) != true)
		{
			bRes = false;
			break;
		}
	}

	if (false == bRes)
	{
		// Deallocate the previous buffers that were successfully been allocated
		// TODO: check if we have deallocate until i or until i-1
		for (j = 0; j < i; ++j)
		{
			int nByteCount = nEveryBucketByteCount; 
			// If this is the last bucket
			if ( j == nNumberOfBuckets-1)
			{
				nByteCount = nLastBucketByteCount;
			}
			DeAllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[j])), nByteCount);
		}

		return false;
	}

	// 1.3 Allocate virtual memory for the relation
	if (AllocateVirtualBuffer((void**)(&psRelBuffer), nHalfBufferLimitedByteCount ) != true)
	{
		bRes = false;		
	}


	// This disk is for reading the relation
	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	
	// The results file name will be wsSortedFileName + 1
	wsRelFileName.append(wsAppend);

	// This disk is for writing the intermediate results (the buckets)
	Disk* psDiskSortedRel	= new Disk(wsRelFileName, eLogical, eWrite);

	// Allocating the buckets map
	SBucketsMapping** asBucketsMap = static_cast<SBucketsMapping**>(malloc(nNumberOfBuckets*sizeof(SBucketsMapping*)));

	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		asBucketsMap[i] = new SBucketsMapping();
	}


	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
		{

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlSoretdOffset	= 0;
			int nSortedBufferIndex		= 0;

			// Calculate the key range and the number of keys in every bucket
			int nKeysRange					= static_cast<int>(fColumDensity*nRelRowsNum);
			int nNumberOfKeysInABucket		= static_cast<int>(floor( static_cast<float>(nKeysRange)/nNumberOfBuckets));
			int nNumberOfKeysInLastBucket	= nNumberOfKeysInABucket;

			if (nEveryBucketByteCount != nLastBucketByteCount)
			{
				// the last bucket is a small one
				float fRel					= (static_cast<float>(nLastBucketByteCount) / ((nNumberOfBuckets-1)*nEveryBucketByteCount));
				nNumberOfKeysInABucket		= static_cast<int>(floor( static_cast<float>((1-fRel)*nKeysRange)/(nNumberOfBuckets-1)));
				nNumberOfKeysInLastBucket	= nKeysRange - (nNumberOfKeysInABucket*(nNumberOfBuckets-1));
			}		

			// Read the relation and divide to the buckets.
			
			for (i = 0; i < nNumberOfBlocksInRel; ++i)
			{
				int nBucketIndex = 0;
				if (true == psDiskRel->DoOverlappedIO(psRelBuffer, nHalfBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					dwlOffset += nHalfBufferLimitedByteCount;

					for (j = 0; j < nEveryBufferRowsCount; ++j)
					{
						int nKey = psRelBuffer[j].GetKey();
						nBucketIndex					=  static_cast<int>(floor( static_cast<float>(psRelBuffer[j].GetKey())/nNumberOfKeysInABucket));
						if (nBucketIndex >= nNumberOfBuckets)
						{
//							printf("this is to the last bucket = index: %d \n", nBucketIndex);
							nBucketIndex = nNumberOfBuckets-1;
						}

						int nIndex									= asBucketsMap[nBucketIndex]->m_nNumOfElems;
						*(psRelBucketsBuffer[nBucketIndex]+nIndex)	= psRelBuffer[j];
						asBucketsMap[nBucketIndex]->m_nNumOfElems++;

						// When a bucket is full - write it to disk.
						// TODO: try to write a whole block (32 sectors) - maybe it doesn't matter - the first one will cost more.

						int nCurrBucketIndex	= asBucketsMap[nBucketIndex]->m_nNumberOfParts;
						int nBucketLimit		= nEveryBucketRowsCount;

						if (nBucketIndex == nNumberOfBuckets -1)
						{
							// this is the last bucket so the number of elements in it is different
							nBucketLimit = nLastBucketRowsCount;

						}

						SBucketsMapping* pCurrBucket = asBucketsMap[nBucketIndex];

						if (nBucketLimit == asBucketsMap[nBucketIndex]->m_nNumOfElems)
						{
							if (NULL != pCurrBucket->m_pCurrPart)
							{
								// We keep the current part of the bucket if there is more than one.
								pCurrBucket = pCurrBucket->m_pCurrPart;
							}

							if (true == psDiskSortedRel->DoOverlappedIO((psRelBucketsBuffer[nBucketIndex]), nBucketLimit*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
							{
								/*printf("Bucket = %d \n", nBucketIndex);*/
								//for (int k = 0; k < nBucketLimit; ++k)
								//{
								//	printf("%d, ", (psRelBucketsBuffer[nBucketIndex] + k)->GetKey());
								//}
								// Keep the offsets of the bucket part in a list.
								pCurrBucket->m_dwPartOffset		= dwlSoretdOffset;
								pCurrBucket->m_pNextPart		= new SBucketsMapping();
								pCurrBucket->m_nBytesCount		= nBucketLimit*sizeof(T);
								
								// Keep the current new part - so that next time we will not have to look for it by
								// iterating over the whole list.
								asBucketsMap[nBucketIndex]->m_pCurrPart	= pCurrBucket->m_pNextPart;
								
								asBucketsMap[nBucketIndex]->m_nNumberOfParts++;
								asBucketsMap[nBucketIndex]->m_nNumOfElems	= 0;
								

								dwlSoretdOffset += nBucketLimit*sizeof(T);
							}
							else
							{
								printf("CSortTests:: failed in psDiskSortedRel->DoOverlappedIO\n");
							}

						}

					}
				}

			}

			// Finish writing the rest of the bucket
			// If there is not full bucket -> wrap it with 0
			for (j = 0; j < nNumberOfBuckets; ++j)
			{
				int nCurrBucketIndex		= asBucketsMap[j]->m_nNumberOfParts;
				int nNumOfElems				= (asBucketsMap[j])->m_nNumOfElems;
				if (nNumOfElems > 0)
				{
					int nSectorRowsCount		= (512/sizeof(T));
					int nNumOfSectors			= ((nNumOfElems / nSectorRowsCount) + 1);
					int nTotalElems				= (nSectorRowsCount * nNumOfSectors);
					int nRestOfElemsToSector	= (nTotalElems - nNumOfElems);
					

					T tempElem;

					for (i = nNumOfElems; i < nTotalElems; ++i)
					{
						*(psRelBucketsBuffer[j]+i) = tempElem;
					}

					SBucketsMapping* pCurrBucket = asBucketsMap[j];
					//while(NULL != pCurrBucket->m_pNextPart)
					//{
					//	pCurrBucket = pCurrBucket->m_pNextPart;
					//}

					if (NULL != pCurrBucket->m_pCurrPart)
					{
						// We keep the current part of the bucket if there is more than one.
						pCurrBucket = pCurrBucket->m_pCurrPart;
					}

					if (true == psDiskSortedRel->DoOverlappedIO((psRelBucketsBuffer[j]), nNumOfSectors*512, dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						/*printf("Bucket = %d \n", j);*/
						// Keep the offsets of the bucket part in a list.
						pCurrBucket->m_dwPartOffset	= dwlSoretdOffset;
						pCurrBucket->m_nBytesCount	= nNumOfSectors*512;
						
						asBucketsMap[j]->m_nNumberOfParts++;
							
						dwlSoretdOffset += nNumOfSectors*512;
					}
					else
					{
						printf("CSortTests:: failed in psDiskSortedRel->DoOverlappedIO\n");
					}
				}
			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

		}

		psDiskRel->CloseDisk();
		psDiskSortedRel->CloseDisk();
	}

	// Closing the disks and deallocating the buffers

	delete(psDiskSortedRel);
	delete(psDiskRel);

	for (j = 0; j < nNumberOfBuckets; ++j)
	{
		int nByteCount = nEveryBucketByteCount; 
		// If this is the last bucket
		if ( j == nNumberOfBuckets-1)
		{
			nByteCount = nLastBucketByteCount;
		}

		DeAllocateVirtualBuffer((void**)(&(psRelBucketsBuffer[j])), nByteCount);
	}



	// This disk is for reading the relation divided to buckets
	psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	
	// This disk is for writing the sorted relation
	 psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

	T* psSortedBuffer = NULL;

	if (AllocateVirtualBuffer((void**)(&psSortedBuffer), nHalfBufferLimitedByteCount ) == true)
	{

		if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
		{
			if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
			{

				LARGE_INTEGER  startTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

				DWORDLONG dwlOffset			= 0;
				DWORDLONG dwlSoretdOffset	= 0;

				// For every bucket - read its parts from the disk
				for (i = 0; i < nNumberOfBuckets; ++i)
				{
					SBucketsMapping* pPart		= NULL;

					pPart = asBucketsMap[i];

					int nBucketRowsCount = 0;
					int nTotalRowsCount = 0;


					// Go to the end of the list and release the memory
					while ((NULL != pPart) && (0 != pPart->m_nBytesCount))
					{
						dwlOffset			= pPart->m_dwPartOffset;
						int nBytesCount		= pPart->m_nBytesCount;
						nBucketRowsCount	= (nBytesCount/sizeof(T));

						if (true != psDiskRel->DoOverlappedIO(&(psRelBuffer[nTotalRowsCount]), nBytesCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
						{
							printf("Problem in reading buckets\n");
							break;
						}
						
						nTotalRowsCount		+= nBucketRowsCount;

						pPart = pPart->m_pNextPart;
					}


					// Sort the whole bucket
					QuickSort2<T>(psRelBuffer, 0, nTotalRowsCount-1);

					int nTotalBytesCount =  nTotalRowsCount*sizeof(T);

					// There might be some 0 to wrap the sector

					if (true != psDiskSortedRel->DoOverlappedIO(psRelBuffer, nTotalBytesCount, dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						printf("Problem in writing buckets\n");
						break;
					}

					dwlSoretdOffset+= nTotalBytesCount;
				}


				LARGE_INTEGER  endTime;
				QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
				rfTimeSec += CalcTime(startTime, endTime);
			}
			psDiskRel->CloseDisk();
			psDiskSortedRel->CloseDisk();
		}
		DeAllocateVirtualBuffer((void**)(&psSortedBuffer), nHalfBufferLimitedByteCount);

	}
			
	// Closing the disks and deallocating the buffers

	delete(psDiskSortedRel);
	delete(psDiskRel);

	DeAllocateVirtualBuffer((void**)(&psRelBuffer), nHalfBufferLimitedByteCount);

	for (i = 0; i < nNumberOfBuckets; ++i)
	{
		SBucketsMapping* pPart		= NULL;
		SBucketsMapping* pNextPart	= NULL;

		pPart = asBucketsMap[i];

		// Go to the end of the list and release the memory
		while (pPart != NULL)
		{
			pNextPart = pPart->m_pNextPart;
			delete(pPart);
			pPart = pNextPart;
		}
		
		asBucketsMap[i] = NULL;
	}

	free(asBucketsMap);

	return bRes;
}

float CSortTests::CalcTime(LARGE_INTEGER startTime, LARGE_INTEGER endTime)
{
	float fTimeMsec = 0;
	fTimeMsec		= static_cast<float>((endTime.QuadPart -startTime.QuadPart) * 1000.0 / m_freq.QuadPart);
	return fTimeMsec;

}



template <class T>
bool CSortTests::FindSortedBlocks(wstring wsRelFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec,  float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum, SSortedPieceInfo** pPartsInfo, int& rnNumberOfSortedBlocks)
{
	bool bRes = true;

	int m_nBufferLimitedRowsCount		= (m_nBufferLimitedByteCount/sizeof(T));
	
	int nReadBufferBytesCount			= (m_nBufferLimitedByteCount/2);

	int nBuffersLimitedRowsNum			= (m_nBufferLimitedRowsCount / sizeof(T));

	int nMaxNumberOfSortedBlocks		= static_cast<int>(floor(nReadBufferBytesCount/512.0f));

	int nCurrNumberOfSortedBlocks		= 1;

	int nBlocksInfoBytesCount			= (nMaxNumberOfSortedBlocks*sizeof(SSortedPieceInfo));

	// TODO: handle the case in which the info is very big and not a part of the memroy size

	//int nBlocksInfoSectorsNum			= static_cast<int>(ceilf(nBlocksInfoBytesCount/512.0f));

	//int nLeftReadBufferBytesCount		= (m_nBufferLimitedByteCount - nBlocksInfoSectorsNum);

	*pPartsInfo							= new SSortedPieceInfo();
	
	SSortedPieceInfo* pCurrPartInfo		= *pPartsInfo;

	// 1. Read the realtion once, gather information about the sorted pieces (offset, size, min number, max number etc)

	T* psRelBuffer				= NULL;

	if (false == AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount))
	{
		rnNumberOfSortedBlocks = -1;
		return false;
	}

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);

	int nNumOfLinesInBlock	= (m_nBufferLimitedByteCount / sizeof(T));
	int nNumOfIterations	= static_cast<int>(ceil(static_cast<float>(nRelByteCount) / m_nBufferLimitedByteCount));

	if (psDiskRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent())
		{		
			DWORDLONG dwlOffset = 0;

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			int nLastKey = 0;

			for (int j = 0; j < nNumOfIterations; ++j)
			{
				dwlOffset = (j * m_nBufferLimitedByteCount);

				// Read a block of the relation
				if (psDiskRel->DoOverlappedIO(psRelBuffer, m_nBufferLimitedByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					for (int k = 0; k < nNumOfLinesInBlock; ++k)
					{
						if (psRelBuffer[k].GetKey() >= nLastKey) 
						{
							// Keep counting the curr sorted block							
							pCurrPartInfo->m_nNumOfElements++;
						}
						else
						{
							// A new sorted blocks started							
							pCurrPartInfo->m_pNextPiece = new SSortedPieceInfo();
							pCurrPartInfo				= pCurrPartInfo->m_pNextPiece;
							
							pCurrPartInfo->m_nNumOfElements++;
							pCurrPartInfo->m_nPartIndex = (j*nNumOfLinesInBlock + k);
							nCurrNumberOfSortedBlocks++;

							if (nCurrNumberOfSortedBlocks > nMaxNumberOfSortedBlocks)
							{
								printf("this relation is not suitable for this kind of sort method\n");
								bRes = false;
								break;
							}

						}

						nLastKey = psRelBuffer[k].GetKey();
					}
				}
			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

			printf("find the blocks took %f msec \n", CalcTime(startTime, endTime));

			psDiskRel->CloseDisk();
		}
	}

	DeAllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount);

	delete psDiskRel;

	rnNumberOfSortedBlocks = nCurrNumberOfSortedBlocks;
	return bRes;
}

// In this function we will run over the relation once, and gather information about the sorted blocks in it.
// Afterwards we will merge these blocks.
template <class T>
bool CSortTests::PiecewiseSortedMerge(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec,  float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum)
{
	bool bRes = true;

	int m_nBufferLimitedRowsCount		= (m_nBufferLimitedByteCount/sizeof(T));
	
	int nReadBufferBytesCount			= (m_nBufferLimitedByteCount/2);

	int nBuffersLimitedRowsNum			= (nReadBufferBytesCount / sizeof(T));

	int nWriteBufferBytesCount			= (m_nBufferLimitedByteCount/2);

	int nWriteBuffersLimitedRowsNum		= (nWriteBufferBytesCount / sizeof(T));

	// TODO: handle the case in which the info is very big and not a part of the memroy size

	SSortedPieceInfo* pPartsInfo		= NULL;
	SSortedPieceInfo* pCurrPartInfo		= NULL;

	// 1. Read the realtion once, gather information about the sorted pieces (offset, size, min number, max number etc)

	int nNumofSortedBlocks = 0;
	bRes = FindSortedBlocks<T>(wsRelFileName, nRelRowsNum, nRelByteCount, rfTimeSec, pfIoTimeMsec, rnReadSectorsNum, rnWrittenSectorsNum, &pPartsInfo, nNumofSortedBlocks);

	if (false == bRes)
	{
		pCurrPartInfo = pPartsInfo;
		//Deallocate the pPartsInfo
		for (int i = 0; i < nNumofSortedBlocks; ++i)
		{
			pCurrPartInfo = pPartsInfo->m_pNextPiece;
			free(pPartsInfo);
			pPartsInfo = pCurrPartInfo;
		}
		return false;
	}

	printf("a\n");
	if (1 == nNumofSortedBlocks)
	{
		printf("the block is already sorted \n");
		return true;
	}

	int nEveryBlockByteCount	= static_cast<int>(floorf(static_cast<float>(nReadBufferBytesCount)/nNumofSortedBlocks));
	int nLastBlockByteCount		= nEveryBlockByteCount;
	int nLastBlockRowsNum		= 0;
	int	nEveryBlockRowsNum		= 0;

	// 2. Merge the relation like in SortInBlocks - handle the unequal sorted blocks size

	// we need N buffers in order to read them together
	T** psRelBuffer				= (T**)malloc(nNumofSortedBlocks*sizeof(T*));
	T* psRelSoretdBuffer		= NULL;

	// TODO: handle when nEveryBufferByteCount is less then a sector
	//////////////////////////////////////////////////////////
	int nEveryBlockSecorsCount = (nEveryBlockByteCount / 512);

	if (0 == nEveryBlockSecorsCount)
	{
		// This will not happen because CreateBlockSortedRelation succeeded
		printf("Shouldn't have happened\n");
	}
	else
	{
		// Every buffer will be aligned to 512  - the last one will be bigger than the others
		nEveryBlockByteCount	= (nEveryBlockSecorsCount * 512);

		int nRemainder			= (nReadBufferBytesCount - (nEveryBlockByteCount*(nNumofSortedBlocks-1)));

		if ( nRemainder > nEveryBlockByteCount)
		{
			nLastBlockByteCount = nRemainder;

			if (0 != (nRemainder%512))
			{
				printf("Problem = 0 != (nRemaider%512) \n");
			}
		}
		else
		{
			nLastBlockByteCount = nEveryBlockByteCount;
		}

		nLastBlockRowsNum		= (nLastBlockByteCount/sizeof(T));

		nEveryBlockRowsNum		= (nEveryBlockByteCount/sizeof(T));

	}
	//////////////////////////////////////////////////////////////

	printf("b\n");
	int i = 0;

	// Allocate virtual memory for the buffers
	for (i = 0; i < nNumofSortedBlocks; ++i)
	{
		int nByteCount = nEveryBlockByteCount;
		if(i == (nNumofSortedBlocks-1))
		{
			nByteCount = nLastBlockByteCount;
		}
		if (AllocateVirtualBuffer((void**)(&(psRelBuffer[i])), nByteCount ) != true)
		{
			bRes = false;
			break;
		}
	}

	if (AllocateVirtualBuffer((void**)(&psRelSoretdBuffer), nWriteBufferBytesCount ) != true)
	{
		bRes = false;		
	}

	if (false == bRes)
	{
		// Deallocate the previous buffers that were successfully been allocated
		// TODO: check if we have deallocate until i or until i-1
		for (int j = 0; j < i; ++j)
		{
			int nByteCount = nEveryBlockByteCount;
			if(j == (nNumofSortedBlocks-1))
			{
				nByteCount = nLastBlockByteCount;
			}
			DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
		}

		printf("false\n");
		return false;
	}

	printf("c\n");
	// This disk is for reding the intermediate results of the sorted buffers.
	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	
	Disk* psDiskSortedRel	= new Disk(wsSortedFileName, eLogical, eWrite);

	printf("d1\n");
	// Allocating the meta-data structure for everyone of the buffers
	SSortedBufferInfo<T>* psBuffers	= (SSortedBufferInfo<T>*)malloc(nNumofSortedBlocks * sizeof(SSortedBufferInfo<T>));

	printf("allocated %d buffers \n", nNumofSortedBlocks);
	if (psDiskRel->OpenDisk() && psDiskSortedRel->OpenDisk())
	{
		printf("d2\n");
		if (psDiskRel->CreateOverlappedEvent() && psDiskSortedRel->CreateOverlappedEvent())
		{
			DWORDLONG dwlOffset			= 0;
			DWORDLONG dwlSoretdOffset	= 0;
			int nSortedBufferIndex		= 0;

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			int nByteCount = nEveryBlockByteCount;
			int nRowsNum   = nEveryBlockRowsNum;

			pCurrPartInfo = pPartsInfo;
			
			printf("d3\n");
			// Read the first part of every buffer
			for (int j = 0; j < nNumofSortedBlocks; ++j)
			{
//				printf("%d\n", j);
//				printf("buffer number %d has %d elements \n", j, pCurrPartInfo->m_nNumOfElements);
				if (j == (nNumofSortedBlocks - 1))
				{
					// This is the last buffer
					nByteCount = min(nLastBlockByteCount, static_cast<int>(pCurrPartInfo->m_nNumOfElements*sizeof(T)));
					nRowsNum   = (nByteCount/sizeof(T));
				}

				// Calculate the offset
				dwlOffset		= static_cast<int>(floor((pCurrPartInfo->m_nPartIndex*sizeof(T))/512.0f)*512);
				int nStartIndex = (pCurrPartInfo->m_nPartIndex % (512/sizeof(T)));
				
				// Read the data
				if (psDiskRel->DoOverlappedIO((psRelBuffer[j]), nByteCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					// Update the meta-data
					psBuffers[j].m_nTotalReadBytes	= nByteCount;

					psBuffers[j].m_nLeftBytesToRead	= ((pCurrPartInfo->m_nNumOfElements - (nRowsNum-nStartIndex))*sizeof(T));

					psBuffers[j].m_nCurrIndex	= nStartIndex;

					psBuffers[j].m_dwOffset		= dwlOffset;	

					psBuffers[j].m_nBytesCount	= nByteCount;

					psBuffers[j].m_nRowsCount	= (nByteCount/sizeof(T));

					// what if there is only one part to read and it is less then a buffer size
					if(0 >= psBuffers[j].m_nLeftBytesToRead)
					{
						psBuffers[j].m_nRowsCount	= nStartIndex + pCurrPartInfo->m_nNumOfElements;
					}

					psBuffers[j].m_bIsFinished = false;

				}
				else
				{
					printf("SortInIterations - Problem - failed reading\n");
					break;
				}

				pCurrPartInfo = pCurrPartInfo->m_pNextPiece;
			}

			bRes = true;

			while(bRes == true)
			{
				bRes = false;

				T* sElem = NULL;

				int nSmallerBufferIndex = 0;
				
				T* sMinElem = NULL;

				int nFirstBuffer = 0;
				// Get the first buffer that we didn't finish reading to be the first smaller one
				for (int j = 0; j < nNumofSortedBlocks; ++j)
				{
//					printf("%d\n", j);
					if (false == psBuffers[j].m_bIsFinished)
					{
						// The first buffer which we didn't finish reading
						sMinElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						nFirstBuffer = j;
						nSmallerBufferIndex = j;
						bRes = true;
						break;
					}

				}				

				// Get the smaller element of whole buffers
				for (int j = nFirstBuffer+1; j < nNumofSortedBlocks; ++j)
				{
					if (false == psBuffers[j].m_bIsFinished)
					{
						sElem = (psRelBuffer[j]) + psBuffers[j].m_nCurrIndex;
						
						if (*sElem < *sMinElem)
						{
							// We have smaller element
							nSmallerBufferIndex = j;
							sMinElem = sElem;
						}

						// We still have buffers that we didn't finish reading
						bRes = true;
					}
				}

//				printf("e\n");

				if (true == bRes)
				{
					// Write the smaller element to the results buffer
					psRelSoretdBuffer[nSortedBufferIndex++] = *(sMinElem);

					// Increase the current index
					psBuffers[nSmallerBufferIndex].m_nCurrIndex++;

					// We have reached to the end of this part - must read another part
					if (psBuffers[nSmallerBufferIndex].m_nRowsCount <= psBuffers[nSmallerBufferIndex].m_nCurrIndex)
					{
						if (0 >= psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead)
						{
							// We have read the whole buffer
							psBuffers[nSmallerBufferIndex].m_bIsFinished = true;
						}
						else
						{
							// We have to read one more part
							dwlOffset = psBuffers[nSmallerBufferIndex].m_dwOffset + psBuffers[nSmallerBufferIndex].m_nBytesCount;

							// Maybe the left bytes to read is smaller than the part size
							int nReadByteCount = min(psBuffers[nSmallerBufferIndex].m_nBytesCount, psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead);

							if (nReadByteCount != psBuffers[nSmallerBufferIndex].m_nBytesCount)
							{
								psBuffers[nSmallerBufferIndex].m_nRowsCount		= (nReadByteCount/sizeof(T));
								// the size must be 512 aligned
								nReadByteCount = static_cast<int>(ceil(psBuffers[nSmallerBufferIndex].m_nBytesCount/512.0f)*512);
								psBuffers[nSmallerBufferIndex].m_nBytesCount	= nReadByteCount;
								
							}
		
							if (psDiskRel->DoOverlappedIO((psRelBuffer[nSmallerBufferIndex]), nReadByteCount,dwlOffset , &rnReadSectorsNum, pfIoTimeMsec))
							{
								psBuffers[nSmallerBufferIndex].m_nTotalReadBytes	+= nReadByteCount;
								psBuffers[nSmallerBufferIndex].m_nCurrIndex			= 0;
								psBuffers[nSmallerBufferIndex].m_dwOffset			= dwlOffset;
								psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead	-= nReadByteCount;

								//if (psBuffers[nSmallerBufferIndex].m_nLeftBytesToRead < 0)
								//{
								//	// this is ok - it means that we finished with that buffer nad 
								//	printf("nnnnnnn\n");
								//}
								psBuffers[nSmallerBufferIndex].m_bIsFinished		= false;
							}
							else
							{
								printf("SortInIterations - Problem - failed reading\n");
								break;
							}
						}
					}
				}

				// If we have reached to the end of the results buffer or we finished reading all blocks 
				//		we have to write the results to the disk
				if ((nWriteBuffersLimitedRowsNum <= nSortedBufferIndex || false == bRes) && (nSortedBufferIndex > 0))
				{
					if (false == psDiskSortedRel->DoOverlappedIO(psRelSoretdBuffer, nSortedBufferIndex*sizeof(T), dwlSoretdOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						printf("SortInIterations - Problem - failed writing\n");
						break;
					}

					dwlSoretdOffset += nSortedBufferIndex*sizeof(T);
					nSortedBufferIndex = 0;
				}

			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

		}
			printf("n\n");

	}
	printf("m\n");
	psDiskRel->CloseDisk();
	psDiskSortedRel->CloseDisk();
	//}

	free(psBuffers);
	delete(psDiskRel);
	delete(psDiskSortedRel);
	//	2.1 A case in which the number of sorted blocks is too big (the (memory / #blocks) < 512)
	//	2.2 A case in which we finished reading one buffer but still reading another one - maybe we will use its buffer.


//	printf("Number of sorted parts is: %d \n", nNumofSortedBlocks);

	// free pPartsInfo
	for (int i = 0; i < nNumofSortedBlocks; ++i)
	{
		pCurrPartInfo = pPartsInfo->m_pNextPiece;
//		printf("Number of elements in part numb %d is = %d \n", i,pPartsInfo->m_nNumOfElements);
		free(pPartsInfo);
		pPartsInfo = pCurrPartInfo;
	}

	// Deallocate the memory for the read-buffers
	for (int j = 0; j < nNumofSortedBlocks; ++j)
	{
		int nByteCount = nEveryBlockByteCount;
		if(j == (nNumofSortedBlocks-1))
		{
			nByteCount = nLastBlockByteCount;
		}

		DeAllocateVirtualBuffer((void**)(&(psRelBuffer[j])), nByteCount);
	}

	// Deallocate the memory for the result buffer
	DeAllocateVirtualBuffer((void**)(&psRelSoretdBuffer), nWriteBufferBytesCount);
	

	return bRes;
}

bool CSortTests::PiecesUnitTests(int nRel2ByteCount)
{	
	SRel2* psRel2Buffer = NULL;
	wstring wsRel2FileNameFlash;
	wstring wsSortedFileNameFlash;
#ifdef __FLASH__
	wsRel2FileNameFlash		= L"\\\\.\\E:\\rel2SortedPartly";
	wsSortedFileNameFlash	= L"\\\\.\\E:\\rel2Sorted";
#else
	wsRel2FileNameFlash		= L"\\\\.\\C:\\rel2SortedPartly";
	wsSortedFileNameFlash	= L"\\\\.\\C:\\rel2Sorted";

#endif // __FALSH_
	
	AllocateVirtualBuffer((void**)(&psRel2Buffer), nRel2ByteCount) ;

	int nNumOfElements		= (nRel2ByteCount/sizeof(SRel2));
	int nNumOfBlocks		= ((m_nBufferLimitedByteCount/2)/512);
	int nNumOfElemsInBlock	= (nNumOfElements/nNumOfBlocks);
	int nIndex				= 0;
	int i					= 0;
	int j					= 0;

	for (i = 0; i < nNumOfBlocks; ++i)
	{
		for (j = 0; j < nNumOfElemsInBlock; ++j)
		{
			psRel2Buffer[nIndex++].m_nc = j;
		}
	}

	psRel2Buffer[17].m_nc = 10000;
	psRel2Buffer[18].m_nc = 20000;


	psRel2Buffer[270].m_nc = 30000;
	psRel2Buffer[271].m_nc = 40000;


	psRel2Buffer[500].m_nc = 50000;
	psRel2Buffer[501].m_nc = 60000;

	psRel2Buffer[1203].m_nc = 70000;
	psRel2Buffer[1204].m_nc = 80000;

	Disk* psDiskRel = new Disk(wsRel2FileNameFlash, eLogical, eWrite);
	if (psDiskRel->OpenDisk() == true)
	{
		if( psDiskRel->CreateOverlappedEvent())
		{
			if (psDiskRel->DoOverlappedIO(psRel2Buffer, nRel2ByteCount))
			{
					cout << "Wrote relation 2 to hard-disk" << endl;
			}
			psDiskRel->CloseDisk();		
		}
	}
	delete psDiskRel;
	
	float fTimeMsec			= 0;
	float fIoTimeMsec		= 0;
	int nReadSectorsNum		= 0;
	int nWrittenSectorsNum	= 0;

	bool bRes = PiecewiseSortedMerge<SRel2>(wsRel2FileNameFlash, wsSortedFileNameFlash, nRel2ByteCount/sizeof(SRel2),
					nRel2ByteCount, fTimeMsec,  &fIoTimeMsec, nReadSectorsNum, nWrittenSectorsNum);

	// Check the results

	psDiskRel	= new Disk(wsSortedFileNameFlash, eLogical, eRead);

	DWORDLONG dwlResOffset	= 0;
	int nResIndex			= 0;
	int nRel2RowsNum		= (nRel2ByteCount/sizeof(SRel2));

	
	if ( psDiskRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent())
		{
			DWORD dwBytesRead			= 0;
			DWORDLONG dwlPrepareOffset	= 0;
			int nResultCounter			= 0;
			int nReadSectors			= 0;

			DWORD dwStart = timeGetTime();


			if (true == psDiskRel->DoOverlappedIO(psRel2Buffer, nRel2ByteCount, dwlResOffset))
			{
				for (int i = 1; i < nRel2RowsNum; ++i)
				{
					if (psRel2Buffer[i-1] > psRel2Buffer[i] && (0!= psRel2Buffer[i].GetKey()))
					{
						printf("not sorted [%d] = %d, [%d] = %d\n", i-1, psRel2Buffer[i-1].GetKey(), i ,  psRel2Buffer[i].GetKey());
					}
//					printf("%d, \n", psRel2Buffer[i-1].GetKey());
				}


				//printf("%d, \n", psRel2Buffer[nRel2RowsNum-1].GetKey());
			}

//			printf("%d, \n", psRel2Buffer[nRel2RowsNum-1].GetKey());
		}
	}
	psDiskRel->CloseDisk();
	delete(psDiskRel);
	DeAllocateVirtualBuffer((void**)(&psRel2Buffer), nRel2ByteCount);

	return bRes;

}


//// For simplicity this method assume that there are no repitions in the input
template <class T>
bool CSortTests::AlmostSortedMerge(wstring wsRelFileName, wstring wsSortedFileName, int nRelRowsNum,
					int nRelByteCount, float& rfTimeSec,  float* pfIoTimeMsec, int& rnReadSectorsNum, int& rnWrittenSectorsNum/*, double fReadRatio /*=0.5*/)
{

	bool bRes = true;

	// 1. Divide the buffers to read and list of unsorted.

	int m_nBufferLimitedRowsCount		= (m_nBufferLimitedByteCount/sizeof(T));
	
	int nReadBufferBytesCount			= static_cast<int>(m_nBufferLimitedByteCount*0.9);

	int nUnSortedBufferBytesCount		= (m_nBufferLimitedByteCount - nReadBufferBytesCount);

	int nBuffersLimitedRowsNum			= (nReadBufferBytesCount / sizeof(T));

	int nMaxNumOfUnsortedElems			= (nUnSortedBufferBytesCount/sizeof(T));

	int nMaxNumberOfSortedBlocks		= static_cast<int>(floor(nReadBufferBytesCount/512.0f));

	int nCurrNumberOfSortedBlocks		= 1;

	int nBlocksInfoBytesCount			= (nMaxNumberOfSortedBlocks*sizeof(SSortedPieceInfo));

	// TODO: Handle the last buffer size is different case


	T* psRelBuffer				= NULL;
	T* psUnsortedBuffer			= NULL; 

	if (false == AllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount))
	{
		return false;
	}

	psUnsortedBuffer = &(psRelBuffer[nBuffersLimitedRowsNum]);

	Disk* psDiskRel			= new Disk(wsRelFileName, eLogical, eRead);
	Disk* psResDiskRel		= new Disk(wsSortedFileName, eLogical, eWrite);

	int nNumOfLinesInBlock	= nBuffersLimitedRowsNum;
	int nNumOfIterations	= static_cast<int>(ceil(static_cast<float>(nRelByteCount) / nReadBufferBytesCount));

	// 2. Read the relation and keep the unsorted elementn in the list
	if (psDiskRel->OpenDisk() && psResDiskRel->OpenDisk())
	{
		if (psDiskRel->CreateOverlappedEvent() && psResDiskRel->CreateOverlappedEvent())
		{		
			DWORDLONG dwlOffset = 0;

			LARGE_INTEGER  startTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

			int nLastKey			= 0;
			int nUnSortedIndex		= 0;

			for (int j = 0; j < nNumOfIterations; ++j)
			{
				dwlOffset = (j * nReadBufferBytesCount);

				// Read a block of the relation
				if (psDiskRel->DoOverlappedIO(psRelBuffer, nReadBufferBytesCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					for (int k = 0; k < nNumOfLinesInBlock; ++k)
					{
						if  ((psRelBuffer[k].GetKey() < nLastKey))
						{
							// Copy the element to the unsorted buffer
							psUnsortedBuffer[nUnSortedIndex++] = psRelBuffer[k];

//							nLastKey = psRelBuffer[k].GetKey();

							if (nMaxNumOfUnsortedElems == nUnSortedIndex)
							{
								// This relation is not fit to this kind of method
								printf("This relation is not fit to this kind of method\n");
								bRes = false;
								break;
							}
						}
						else if ((psRelBuffer[k].GetKey() > nLastKey) && 
									((k < nNumOfLinesInBlock-1)&& 
									(psRelBuffer[k+1].GetKey() > nLastKey)&& 
									(psRelBuffer[k].GetKey() > psRelBuffer[k+1].GetKey())))
						{
														// Copy the element to the unsorted buffer
							psUnsortedBuffer[nUnSortedIndex++] = psRelBuffer[k];

//							nLastKey = psRelBuffer[k].GetKey();

							if (nMaxNumOfUnsortedElems == nUnSortedIndex)
							{
								// This relation is not fit to this kind of method
								printf("This relation is not fit to this kind of method\n");
								bRes = false;
								break;
							}

						}
						else
						{
							// Everything is ok - the buffer is sorted
							nLastKey = psRelBuffer[k].GetKey();
						}
					}
					if (false == bRes)
					{
						break;
					}
				}
			}

			LARGE_INTEGER  endTime;
			QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
			rfTimeSec += CalcTime(startTime, endTime);

			printf("find the unsorted elements took %f msec \n", CalcTime(startTime, endTime));


			// 3. Sort the unsorted list

			QuickSort2<T>(psUnsortedBuffer, 0, nUnSortedIndex-1);


			// 4. Read again the relation and insert to it the relevant unsorted elements
			dwlOffset				= 0;
			DWORDLONG dwlResOffset	= 0;

			T tempElem;
			tempElem.m_nc = (2*nRelByteCount);
			for (int j = 0; j < nNumOfIterations; ++j)
			{
				dwlOffset = (j * nReadBufferBytesCount);

				// Read a block of the relation
				if (psDiskRel->DoOverlappedIO(psRelBuffer, nReadBufferBytesCount, dwlOffset, &rnReadSectorsNum, pfIoTimeMsec))
				{
					T* pTemp = psRelBuffer; 
					int k = 0;
					int nMinUnsortedIndex = 0;
					// TODO: handle the case in which we don't have to sort at all
					for (k = 0; k < nNumOfLinesInBlock; ++k)
					{
						if ((nUnSortedIndex == 0) || (nUnSortedIndex == nMinUnsortedIndex))
						{
							break;
						}
						int nMin			= pTemp[k].GetKey();
						int nUnsortedMin	= psUnsortedBuffer[nMinUnsortedIndex].GetKey();
						if ((nMin > nUnsortedMin ) ||
							((k >0) && (nMin < pTemp[k-1].GetKey())))
						{
							printf("Switched bwtween %d, %d \n", pTemp[k].GetKey(), psUnsortedBuffer[nMinUnsortedIndex].GetKey());
							pTemp[k] = psUnsortedBuffer[nMinUnsortedIndex];
							// this element is in the unsorted list (or have been already written to disk)
							// Increment the buffer pointer
							psUnsortedBuffer[nMinUnsortedIndex++] = tempElem;
						}
						//else
						//{
						//	break;
						//}
					}
					
					//nUnSortedIndex -= (k);
					
					if (0 != nUnSortedIndex)
					{
						QuickSort2<T>(pTemp, 0, nUnSortedIndex + nBuffersLimitedRowsNum-1);
						nUnSortedIndex -= nMinUnsortedIndex;
					}

					

					if ( true == psResDiskRel->DoOverlappedIO(psRelBuffer, nReadBufferBytesCount, dwlResOffset, &rnWrittenSectorsNum, pfIoTimeMsec))
					{
						dwlResOffset += nReadBufferBytesCount;
					}

					// Copy the left of the unsorted list to its place
//					memcpy(psUnsortedBuffer, &(psUnsortedBuffer[k]), nUnSortedIndex);
				}
			}

			psDiskRel->CloseDisk();

			psResDiskRel->CloseDisk();
		}
	}

	DeAllocateVirtualBuffer((void**)(&psRelBuffer), m_nBufferLimitedByteCount);

	delete psDiskRel;
	delete psResDiskRel;

	return bRes;

}


template <class T>
int CSortTests::Merge2SortedArrays(T* pBuffer, int nNumOfElems1, int nNumOfElems2, int nMaxSize)
{
	int nHalf = static_cast<int>(ceil(nMaxSize/2.0f));
	int nRes = 0;

	if ((nNumOfElems1 < nHalf && nNumOfElems2 < nHalf) || (0 == nNumOfElems1))
	{
		// both parts are less than half - sort them together
		QuickSort2<T>(pBuffer, 0, nNumOfElems1+nNumOfElems2-1, false);
		nRes = nNumOfElems1+nNumOfElems2;
	}
	else
	{
		// Assunption: the two buffers have equal sizes
		T* p1 = pBuffer;
		T* p2 = &(pBuffer[nNumOfElems1]);

		//LARGE_INTEGER  startTime;
		//QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

		// 0. Sort p2
		QuickSort2<T>(p2, 0, nNumOfElems2-1, false);

		//LARGE_INTEGER  endTime;
		//QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
		//float fTimeSec = CalcTime(startTime, endTime);

		//printf("case 0 took %f [msec] \n", fTimeSec);

#ifdef __SORT_DEBUG__
		printf("\nAfter Sorting \n P1 =");
		for (int i = 0; i <nNumOfElems1; ++i)
		{
			printf("%d, ", p1[i]);
		}

		printf("\n P2=");

		for (int i = 0; i <nNumOfElems2; ++i)
		{
			printf("%d, ", p2[i]);
		}

#endif // __SORT_DEBUG__

		// 1. Switch elements between p1 and p2 (from the begining) so in p1
		//		will be the smaller elements and in p2 the biggers

		int nP1Index = 0;
		int nP2Index = 0;

		while(nP1Index < nNumOfElems1 && nP2Index < nNumOfElems2)
		{
			if (p1[nP1Index] > p2[nP2Index])
			{
				// switch between them
				T temp			= p2[nP2Index];
				p2[nP2Index]	= p1[nP1Index];
				p1[nP1Index]	= temp;
			}
			nP1Index++;
			nP2Index++;
		}

#ifdef __SORT_DEBUG__
		printf("\nAfter Switching \n P1 =");
		for (int i = 0; i <nNumOfElems1; ++i)
		{
			printf("%d, ", p1[i]);
		}

		printf("\n P2=");

		for (int i = 0; i <nNumOfElems2; ++i)
		{
			printf("%d, ", p2[i]);
		}

#endif// __SORT_DEBUG__

		// Now in the first half of p1 there are the smallest elements in the buffer - we will keep them
		//	in the second half of p2 there are the bigest elements in the buffer - we will throw them away.

		// 2. Get the next half smallest elements

		nP1Index			= nNumOfElems1-1; // we start from the end
		int nMiddleIndex	= static_cast<int>(floor(nMaxSize/2.0f))-1;
		nP2Index			= min(nMiddleIndex, nNumOfElems2-1); // we start from the middle 

		if (nP1Index < nMiddleIndex)
		{
			// P1 is smaller than M/2 so we will take it as is and will take the 
			// rest of the elements from P2

			int nElems = min(nNumOfElems1+nNumOfElems2, nMaxSize);
			// We will just sort the buffer and this is the result
			QuickSort2<T>(p1, 0, nElems-1, false);
			nRes = nElems;
		}
		else if (nNumOfElems1+nNumOfElems2 < nMaxSize)
		{
			// the total number of elements is less than nMaxSize but p1 is bigger that nHalf

			// Copy p2 to the next place of p1 and sort it
			memcpy(&(p1[nP1Index+1]), &(p2[0]),(nP2Index+1)*sizeof(T));

			QuickSort2<T>(p1, 0, nNumOfElems1+nNumOfElems2-1, false);

			nRes = nNumOfElems1+nNumOfElems2;
		}
		else
		{
			int nP1Missing	= (nMaxSize - nNumOfElems1);
			int nP2Missing	= (nMaxSize - nNumOfElems2);

			int nNextHalfSize	= 0;

			nNextHalfSize = nMiddleIndex+1;

			if (nP2Missing > nMiddleIndex)
			{
				// more than half is missing
				nP2Missing		= (nP2Missing % nHalf);
				nNextHalfSize	-= (nP1Missing + nP2Missing);
			}
			else
			{
				// a part of the second half of p2 is missing - we still don't want
				// the whole second part of p2. we will start checking in the middle,

				// We want to replace the P1 missing elements
				nNextHalfSize -= nP1Missing;
			}

			while(nP1Index > nMiddleIndex && nP2Index >= 0 && nNextHalfSize > 0)
			{
				if (p1[nP1Index] > p2[nP2Index])
				{
					// we have to throw this element away
					nP1Index--;
				}
				else
				{
					nP2Index--;
				}
				nNextHalfSize--;
			}

#ifdef __SORT_DEBUG__
			printf("\nAfter Switching \n P1 =");
			for (int i = 0; i <nP1Index+1; ++i)
			{
				printf("%d, ", p1[i]);
			}

			printf("\n P2=");

			for (int i = 0; i <nP2Index+1; ++i)
			{
				printf("%d, ", p2[i]);
			}

#endif // __SORT_DEBUG__

			//3. Copy the rest of p2 to p1
			memcpy(&(p1[nP1Index+1]), &(p2[0]),(nP2Index+1)*sizeof(T)); 

#ifdef __SORT_DEBUG__
			printf("\nAfter Throwing Garbage \n P1 =");
			for (int i = 0; i <nMaxSize; ++i)
			{
				printf("%d, ", p1[i]);
			}

			printf("\n");

#endif // __SORT_DEBUG__

			// 4. Sort p1 and this is the final result

			QuickSort2<T>(p1, 0, nMaxSize-1, false);

			nRes = nMaxSize;
		}
	}

#ifdef __SORT_DEBUG__
	printf("After Sorting \n P1 =");
	for (int i = 0; i <nRes; ++i)
	{
		printf("%d, ", pBuffer[i]);
	}
#endif // __SORT_DEBUG__

	return nRes;

}



void CSortTests::Merge2SortedArraysTest(int nRowsNum)
{	


	//SRel2* p1 = static_cast<SRel2*>(malloc(nRowsNum*sizeof(SRel2)));
	//SRel2* p2 = static_cast<SRel2*>(malloc(nRowsNum*sizeof(SRel2)));

	//// Create the relation
	//for (int i = 0; i < nRowsNum; ++i)
	//{
	//	float fd		= static_cast<float>(rand()) / RAND_MAX;
	//	int nResult		= static_cast<int>(fd * nRowsNum);

	//	p1[i].m_nc = nResult;
	//	p2[i].m_nc = nResult;
	//}

	//// Sort Half of the relation
	//QuickSort2<SRel2>(p1, 0, nRowsNum/2-1-1);
	//QuickSort2<SRel2>(p2, 0, nRowsNum/2-1-1);

	//// Check time of QuickSort2
	//LARGE_INTEGER  startTime;
	//QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

	//QuickSort2<SRel2>(p2, 0, nRowsNum-1-1);

	//LARGE_INTEGER  endTime;
	//QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
	//float fTimeSec1 = CalcTime(startTime, endTime);

	//// Check time of Merge2SortedArrays

	//QueryPerformanceCounter((LARGE_INTEGER *)&startTime);

	//Merge2SortedArrays<SRel2>(p1, nRowsNum/2-1, nRowsNum/2, nRowsNum/2);

	//QueryPerformanceCounter((LARGE_INTEGER *)&endTime);
	//float fTimeSec2 = CalcTime(startTime, endTime);

	//printf("Quick Sort took %f [msec] and Merge2SortedArrays took %f [msec]\n", fTimeSec1, fTimeSec2); 

	//for (int i = 0; i < nRowsNum/2; ++i)
	//{
	//	if (p1[i].GetKey() != p2[i].GetKey())
	//	{
	//		printf("Problem = p1[%d] = %d - p2[%d] = %d\n",i,  p1[i].GetKey(), i , p2[i].GetKey()); 
	//	}
	//}

	//return;
	int p[24] = {1,3,5,7,8,9,10, 14, 20,  2,4,11,12};

	Merge2SortedArrays<int>(p, 9, 4, 10);

	//int p1[100];
	//int p2[100];

	//for(int i = 0; i < 20; ++i)
	//{
	//	p1[i] = (rand()%1000)+200;
	//	p2[i] = p1[i];
	//}

	//QuickSort2<int>(p1, 0, 8-1);
	//QuickSort2<int>(p2, 0, 8-1);

	//QuickSort2<int>(p2, 0, 20-1-3);
	//
	//Merge2SortedArrays<int>(p1, 8, 9, 10);

	//for (int i = 0; i < 10; ++i)
	//{
	//	if (p1[i] != p2[i])
	//	{
	//		printf("Problem = p1[%d] = %d - p2[%d] = %d\n",i,  p1[i], i , p2[i]); 
	//	}
	//}

	//// when both parts are more than half but less than whole
	//int p1[16] = {1,3,6,7,8,9,11, 2,4,12,10,5,17,15,14,13 };
	//Merge2SortedArrays<int>(p1, 7, 9, 10);

	//// when p1 is less than half and p2 is more thatn half and less than whole
	//int p2[] = {1,3,6,7,		2,4,5,10,12,13,14,15};

	//Merge2SortedArrays<int>(p2, 4, 8, 10);

	//// when p1 and p2 are less than half
	//int p3[] = {1,3,6,7,		2,4,5};

	//Merge2SortedArrays<int>(p3, 4, 3, 10);

	//// when p2 is less than half and p1 is more than half but less than whole
	//int p4[] = {2,4,5,10,12,13,14,15,		1,3,6,7};

	//Merge2SortedArrays<int>(p4, 8, 4, 10);


}